Help us shape the future of Go! Take the 2026 Go Developer Survey

iop

package
v1.6.5 Latest Latest
Warning

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

 Go to latest Published: Oct 8, 2026 License: GPL-3.0 Imports: 108 Imported by: 5

Documentation

Index

Constants

View Source 
const (
	DuckDbDefaultMaxLineSize = 2000000
	DuckDbLargeMaxLineSize   = 256 * 1024 * 1024
)
View Source 
const ArrowLaneBuffer = 8

ArrowLaneBuffer is the record channel depth of a RecordStream. It is the lane's backpressure: the producer blocks once this many records wait for the consumer.

Variables

View Source 
var (
	// RemoveTrailingDecZeros removes the trailing zeros in CastToString
	RemoveTrailingDecZeros = false
	SampleSize             = 900
)
View Source 
var (
	DuckDbVersion     = "1.5.5"
	DuckDbVersionMD   = "1.5.5"
	DuckDbUseTempFile = false

	DuckDbURISeparator = "|-|+|"
)
View Source 
var (
	TransformsLegacyMap = map[string]TransformLegacy{}

	GlobalFunctionMap map[string]goval.ExpressionFunction

	GetTransformFunction = func(string) goval.ExpressionFunction {
		return nil
	}

	FunctionToTransform = func(name string, f goval.ExpressionFunction, params ...any) TransformLegacy {
		return TransformLegacy{
			Name: name,
			Func: func(sp *StreamProcessor, vals ...any) (any, error) {
				if len(params) > 0 {
					vals = append(vals, params...)
				}
				val, err := f(vals...)
				return val, err
			},
		}
	}

	LocalConnections = cmap.New[map[string]any]()

	// ResolveSecret resolves one secret reference for the secret() function.
	// The connection package sets it, so iop does not import the resolver setup.
	ResolveSecret func(ref string) (string, error)
)
View Source 
var (
	TransformDecodeLatin1 = TransformLegacy{
		Name: EncodingLatin1.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeISO8859_1, val)
			return newVal, err
		},
	}

	TransformDecodeLatin5 = TransformLegacy{
		Name: EncodingLatin5.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeISO8859_5, val)
			return newVal, err
		},
	}

	TransformDecodeLatin9 = TransformLegacy{
		Name: EncodingLatin9.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeISO8859_15, val)
			return newVal, err
		},
	}

	TransformDecodeUtf8 = TransformLegacy{
		Name: EncodingUtf8.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeUTF8, val)
			return newVal, err
		},
	}

	TransformDecodeUtf8Bom = TransformLegacy{
		Name: EncodingUtf8Bom.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeUTF8BOM, val)
			return newVal, err
		},
	}

	TransformDecodeUtf16 = TransformLegacy{
		Name: EncodingUtf16.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeUTF16, val)
			return newVal, err
		},
	}

	TransformDecodeWindows1250 = TransformLegacy{
		Name: EncodingWindows1250.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeWindows1250, val)
			return newVal, err
		},
	}

	TransformDecodeWindows1252 = TransformLegacy{
		Name: EncodingWindows1252.DecodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.DecodeWindows1252, val)
			return newVal, err
		},
	}

	TransformDuckdbListToText = TransformLegacy{
		Name: "duckdb_list_to_text",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.duckDbListAsText(val), nil
		},
	}

	TransformEncodeLatin1 = TransformLegacy{
		Name: EncodingLatin1.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeISO8859_1, val)
			return newVal, err
		},
	}

	TransformEncodeLatin5 = TransformLegacy{
		Name: EncodingLatin5.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeISO8859_5, val)
			return newVal, err
		},
	}

	TransformEncodeLatin9 = TransformLegacy{
		Name: EncodingLatin9.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeISO8859_15, val)
			return newVal, err
		},
	}

	TransformEncodeUtf8 = TransformLegacy{
		Name: EncodingUtf8.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return fmt.Sprintf("%q", val), nil
			newVal, _, err := transform.String(sp.transformers.EncodeUTF8, val)
			return newVal, err
		},
	}

	TransformEncodeUtf8Bom = TransformLegacy{
		Name: EncodingUtf8Bom.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeUTF8BOM, val)
			return newVal, err
		},
	}

	TransformEncodeUtf16 = TransformLegacy{
		Name: EncodingUtf16.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeUTF16, val)
			return newVal, err
		},
	}

	TransformEncodeWindows1250 = TransformLegacy{
		Name: EncodingWindows1250.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeWindows1250, val)
			return newVal, err
		},
	}

	TransformEncodeWindows1252 = TransformLegacy{
		Name: EncodingWindows1252.EncodeString(),
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			newVal, _, err := transform.String(sp.transformers.EncodeWindows1252, val)
			return newVal, err
		},
	}

	TransformHashMd5 = TransformLegacy{
		Name: "hash_md5",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return g.MD5(val), nil
		},
	}

	TransformHashSha256 = TransformLegacy{
		Name: "hash_sha256",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.SHA256(val), nil
		},
	}

	TransformHashSha512 = TransformLegacy{
		Name: "hash_sha512",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.SHA512(val), nil
		},
	}

	TransformParseBit = TransformLegacy{
		Name: "parse_bit",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ParseBit(sp, val)
		},
	}

	TransformBinaryToDecimal = TransformLegacy{
		Name: "binary_to_decimal",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.BinaryToDecimal(sp, val)
		},
	}

	TransformBinaryToHex = TransformLegacy{
		Name: "binary_to_hex",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.BinaryToHex(val), nil
		},
	}

	TransformParseFix = TransformLegacy{
		Name: "parse_fix",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ParseFIX(sp, val)
		},
	}

	TransformParseUuid = TransformLegacy{
		Name: "parse_uuid",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ParseUUID(sp, val)
		},
	}

	TransformParseMsUuid = TransformLegacy{
		Name: "parse_ms_uuid",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ParseMsUUID(sp, val)
		},
	}

	TransformReplace0x00 = TransformLegacy{
		Name: "replace_0x00",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.Replace0x00(sp, val)
		},
	}

	TransformReplaceAccents = TransformLegacy{
		Name: "replace_accents",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ReplaceAccents(val), nil
		},
	}

	TransformReplaceNonPrintable = TransformLegacy{
		Name: "replace_non_printable",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return Transforms.ReplaceNonPrintable(val), nil
		},
	}

	TransformTrimSpace = TransformLegacy{
		Name: "trim_space",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return strings.TrimSpace(val), nil
		},
	}

	TransformLower = TransformLegacy{
		Name: "lower",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return strings.ToLower(val), nil
		},
	}

	TransformUpper = TransformLegacy{
		Name: "upper",
		FuncString: func(sp *StreamProcessor, val string) (string, error) {
			return strings.ToUpper(val), nil
		},
	}

	// used as lookup, cannot return null since is not pointer
	TransformEmptyAsNull = TransformLegacy{
		Name: "empty_as_null",
		Func: func(sp *StreamProcessor, vals ...any) (any, error) {
			if len(vals) == 0 {
				return nil, nil
			}
			if vals[0] == "" {
				return nil, nil
			}
			return vals[0], nil
		},
	}
)
View Source 
var AllCompressorType = []struct {
	Value  CompressorType
	TSName string
}{
	{AutoCompressorType, "AutoCompressorType"},
	{NoneCompressorType, "NoneCompressorType"},
	{ZipCompressorType, "ZipCompressorType"},
	{GzipCompressorType, "GzipCompressorType"},
	{SnappyCompressorType, "SnappyCompressorType"},
	{ZStandardCompressorType, "ZStandardCompressorType"},
}
View Source 
var ColMetaKeys = []ColMetaKey{
	ColMetaAutoIncrement, ColMetaIdentitySeed, ColMetaIdentityIncr,
	ColMetaNullable, ColMetaDefaultValue, ColMetaIsPrimaryKey, ColMetaUnique,
	ColMetaIndex, ColMetaDDLExplicit, ColMetaForeignKey,
	ColMetaCheckConstraint, ColMetaDescription,
}

ColMetaKeys is a list of all column metadata keys

View Source 
var DefaultExprs = []DefaultExpr{
	DefaultExprCurrentTimestamp, DefaultExprCurrentTimestampUTC,
	DefaultExprCurrentDate, DefaultExprCurrentTime,
	DefaultExprUUID, DefaultExprUUIDSequential,
	DefaultExprTrue, DefaultExprFalse, DefaultExprNull,
}

DefaultExprs is a list of all standard default expressions

View Source 
var KeyTypes = []KeyType{AggregateKey, ClusterKey, DuplicateKey, HashKey, IndexKey, PartitionKey, PrimaryKey, SortKey, UniqueKey, UpdateKey}
View Source 
var NewTransform = func(t []map[string]string, _ *StreamProcessor) Transform {
	return nil
}
View Source 
var ParseStageTransforms = func(payload any) ([]map[string]string, error) {
	return nil, g.Error("please use the official sling-cli release for using transforms")
}
View Source 
var PartitionLevelsAscending = []PartitionLevel{
	PartitionLevelSecond,
	PartitionLevelMinute,
	PartitionLevelHour,
	PartitionLevelDay,
	PartitionLevelWeek,
	PartitionLevelMonth,
	PartitionLevelYearMonth,
	PartitionLevelYear,
}
View Source 
var PartitionLevelsDescending = []PartitionLevel{
	PartitionLevelYear,
	PartitionLevelYearMonth,
	PartitionLevelMonth,
	PartitionLevelWeek,
	PartitionLevelDay,
	PartitionLevelHour,
	PartitionLevelMinute,
	PartitionLevelSecond,
}
View Source 
var QueuePollInterval = 50 * time.Millisecond

QueuePollInterval is how often a tailing reader re-checks the file at EOF.

View Source 
var TimeLevelAscending = []TimeLevel{
	TimeLevelSecond,
	TimeLevelMinute,
	TimeLevelHour12,
	TimeLevelHour24,
	TimeLevelDayOfYear,
	TimeLevelDay,
	TimeLevelMonth,
	TimeLevelMonthName,
	TimeLevelYearShort,
	TimeLevelYear,
}
View Source 
var TimeLevelDescending = []TimeLevel{
	TimeLevelYear,
	TimeLevelYearShort,
	TimeLevelMonthName,
	TimeLevelMonth,
	TimeLevelDay,
	TimeLevelDayOfYear,
	TimeLevelHour24,
	TimeLevelHour12,
	TimeLevelMinute,
	TimeLevelSecond,
}
View Source 
var Transforms transformsNS

Functions

func AppendToBuilder added in v1.4.10 

func AppendToBuilder(builder array.Builder, col *Column, val interface{})

func ApplySelect added in v1.5.1 

func ApplySelect(fields []string, selectExprs []string) (newFields []string, err error)

ApplySelect filters, renames, and reorders fields per the select grammar: 

"field" include · "-field" exclude · "field as new" rename
"*" all-not-pinned · "prefix*"/"*suffix" glob include · "-prefix*"/"-*suffix" glob exclude

Output follows expression order; `*` and globs expand in source order and skip names pinned elsewhere in the list (so `[id, *, created_at]` keeps `created_at` at the tail). Matching is case-insensitive. In explicit mode (no `*`), an exact include of a missing source field is a hard error.

func ApplySelectExprs added in v1.5.19 

func ApplySelectExprs(fields []string, selectExprs []string) (newFields []string, err error)

ApplySelectExprs is ApplySelect but emits "field as alias" in the output instead of the post-rename name — for SQL builders that need the source column name before the AS keyword. Semantics otherwise identical.

func ArrowIPCFileSchema added in v1.6.4 

func ArrowIPCFileSchema(file *os.File) (schema *arrow.Schema, err error)

ArrowIPCFileSchema returns the schema of an Arrow IPC file from its footer and rewinds the file, so the same handle can be read afterwards. Only the footer is read.

func AutoDecompress 

func AutoDecompress(reader io.Reader) (gReader io.Reader, err error)

AutoDecompress auto detects compression to decompress. Otherwise return same reader

func CleanHeaderRow 

func CleanHeaderRow(header []string) []string

CleanHeaderRow cleans the header row from incompatible characters

func CleanName 

func CleanName(name string) (newName string)

func CloseQueues added in v1.4.23 

func CloseQueues()

func ColumnsToArrowSchema added in v1.4.10 

func ColumnsToArrowSchema(columns Columns) *arrow.Schema

func CompareColumns 

func CompareColumns(columns1 Columns, columns2 Columns) (reshape bool, err error)

CompareColumns compared two columns to see if there are similar

func CreateDummyFields 

func CreateDummyFields(numCols int) (cols []string)

CreateDummyFields creates dummy columns for csvs with no header row

func DecodeJSONIfBase64 added in v1.5.1 

func DecodeJSONIfBase64(jsonBody string) (string, error)

DecodeJSONIfBase64 returns jsonBody as-is when it's already valid JSON, otherwise tries base64-decode and returns the decoded JSON. Falls back to the original on any failure (malformed JSON passes through unchanged).

func ExtractPartitionTimeValue added in v1.3.6 

func ExtractPartitionTimeValue(mask, path string) (timestamp time.Time, err error)

ExtractPartitionTimeValue extracts from partition time value from the path with mask `data/{YYYY}/{MM}/{DD}` and path `data/2024/12/21`, the returned timestamp should be 2024-12-21.

func FirstObjectKeysInOrder added in v1.5.19 

func FirstObjectKeysInOrder(b []byte) ([]string, error)

FirstObjectKeysInOrder returns the source-order keys of the first top-level object (or the first array element). Recovers what json.Unmarshal into map[string]any loses. Returns nil for non-object roots.

func FlattenRecord added in v1.5.20 

func FlattenRecord(rec map[string]any, depth int) map[string]any

FlattenRecord flattens a record's nested objects/arrays into `__`-joined keys, matching the jsonStream's flattening (Delimiter "__", Safe). depth is the max flatten level (use 1 for the API default); depth < 0 returns the record unchanged. This lets callers apply a `select` against flattened field names (e.g. `owner__login`) the same way the stream does downstream.

func FormatValue added in v1.2.21 

func FormatValue(val any, columnType ColumnType, connType dbio.Type) (newVal string)

FormatValue format as sql expression (adds quotes)

func GeneratePartURIsFromRange added in v1.3.6 

func GeneratePartURIsFromRange(mask, updateKey string, start, end time.Time) (uris []string, err error)

GeneratePartURIsFromRange generates all the possible URIs given a start/end range. Must first determine the lowest time resolution

func GetISO8601DateMap 

func GetISO8601DateMap(t time.Time) map[string]any

GetISO8601DateMap return a map of date parts for string formatting

func GetLowestPartTimeUnit added in v1.3.6 

func GetLowestPartTimeUnit(mask string) (time.Duration, error)

GetLowestPartTimeUnit loops though possible TimeLevel or PartitionLevel values and returns the lowest time.Duration unit

func GetPartitionDateMap added in v1.3.6 

func GetPartitionDateMap(partKeyPrefix string, timestamp time.Time) map[string]any

func GetValueFromArrowArray added in v1.4.10 

func GetValueFromArrowArray(arr arrow.Array, idx int) any

GetValueFromArrowArray extracts a value from an arrow array at the given index

func IsDuckDbProcDeath added in v1.6.4 

func IsDuckDbProcDeath(err error) bool

IsDuckDbProcDeath reports whether err comes from a duckdb sidecar that died mid-query.

func IsDummy 

func IsDummy(columns []Column) bool

IsDummy returns true if the columns are injected by CreateDummyFields

func IsStandardDefault added in v1.5.7 

func IsStandardDefault(value string) bool

IsStandardDefault checks if a value matches a known generalized default

func IsZipBytes added in v1.5.7 

func IsZipBytes(data []byte) bool

isZipBytes checks for the zip file magic number (PK\x03\x04)

func Iso8601ToGoLayout 

func Iso8601ToGoLayout(dateFormat string) (goDateFormat string)

https://www.w3.org/QA/Tips/iso-date https://www.w3.org/TR/NOTE-datetime https://www.iso.org/iso-8601-date-and-time-format.html

func JqCompile added in v1.5.7 

func JqCompile(expr string) (*gojq.Code, error)

JqCompile returns a compiled jq code for the given expression, using a cache.

func JqRun added in v1.5.7 

func JqRun(expr string, input any) ([]any, error)

JqRun compiles and runs a jq expression against input, returning all results.

func MakeAwsConfig added in v1.4.11 

func MakeAwsConfig(ctx context.Context, props map[string]string) (cfg aws.Config, err error)

func MakeDecNumScale added in v1.1.6 

func MakeDecNumScale(scale int) *big.Rat

func MakeIndexName added in v1.5.20 

func MakeIndexName(tableName string, parts []string, maxLength int) string

MakeIndexName builds a deterministic index name from a table name and the index's column/expression names. When the result exceeds maxLength (the dialect's identifier limit), it truncates and appends a short stable hash suffix so distinct auto-named indexes never collide on a shared prefix.

func MakeRowsChan 

func MakeRowsChan() chan []any

MakeRowsChan returns a buffered channel with default size

func MatchedPartitionMask added in v1.3.6 

func MatchedPartitionMask(mask, path string) (matches bool)

MatchedPartitionMask determines if the mask and the path have the same partition structure

func MatchesSelectGlob added in v1.5.5 

func MatchesSelectGlob(name, pattern string) bool

MatchesSelectGlob matches name against a simple glob (prefix*, *suffix, *middle*, prefix*suffix). Both inputs must already be lowercased.

func NewJSONStream 

func NewJSONStream(ds *Datastream, decoder decoderLike, flatten int, jmespath string, jq string) *jsonStream

func NewXMLDecoder added in v1.4.20 

func NewXMLDecoder(reader io.Reader) *xmlDecoder

func OpenTunnelProxy added in v1.5.9 

func OpenTunnelProxy(proxyURL, tgtHost string, tgtPort int) (localPort int, err error)

OpenTunnelProxy forwards traffic through the SOCKS5 proxy

func OpenTunnelSSH added in v1.2.15 

func OpenTunnelSSH(tgtHost string, tgtPort int, tunnelURL, privateKey, passphrase string, options SSHOptions) (localPort int, err error)

func ParseSelectExpr added in v1.5.5 

func ParseSelectExpr(expr string) (field string, newName string, exclude bool, err error)

ParseSelectExpr splits one select expression into (field, newName, exclude). Recognizes "-field" exclusion and case-insensitive " as " rename; the latter requires surrounding spaces so column names like "has_value" don't match.

func Row 

func Row(vals ...any) []any

Row is a row

func ScanCarrRet 

func ScanCarrRet(data []byte, atEOF bool) (advance int, token []byte, err error)

ScanCarrRet removes the \r runes that are without \n rightafter

func SetSampleSize added in v1.4.21 

func SetSampleSize()

func StringToDecimalByteArray added in v1.1.6 

func StringToDecimalByteArray(s string, numSca *big.Rat, pType parquet.Type, length int) []byte

func StripSQLComments added in v1.4.6 

func StripSQLComments(sql string) (string, error)

StripSQLComments removes all SQL comments (-- or /* */) from the provided SQL string

func TotalRecordSize added in v1.6.4 

func TotalRecordSize(rec arrow.RecordBatch) int64

TotalRecordSize returns the in-memory size of a record's buffers.

func Unzip 

func Unzip(src string, dest string) (nodes []map[string]any, err error)

Unzip will decompress a zip archive, moving all files and folders within the zip file (parameter 1) to an output directory (parameter 2).

Types

type ArrowLane added in v1.6.4 

type ArrowLane interface {
	// CastSupported reports whether every value of `from` survives a cast to
	// `to` without loss. The reason is set for every decline.
	CastSupported(from, to arrow.DataType) (ok bool, reason string)
	// Normalize returns a record in the `to` schema, casting only the fields
	// whose type differs.
	Normalize(rec arrow.RecordBatch, to *arrow.Schema) (arrow.RecordBatch, error)
	// Project returns a record with the names and order of cols.
	Project(rec arrow.RecordBatch, cols Columns) (arrow.RecordBatch, error)
	// MaxOf returns the maximum value of an array as unix micro (or as the
	// int64 value for integer types). ok is false when the type is not
	// tracked or the array holds no non-null value.
	MaxOf(arr arrow.Array) (val int64, ok bool)
	// NewTransform returns the evaluator for a stage list. One per stream: it
	// holds the column types the first record fixed, as the row path's
	// transform does. A stage the lane cannot evaluate is an error: the gate
	// classifies the stages first, so this only fires on a wiring bug.
	NewTransform(stages []map[string]string, sp *StreamProcessor) (RecordTransform, error)
	// ClassifyTransform returns the reason the first stage the lane cannot
	// evaluate declines, or "" when it can evaluate every stage. cols may be
	// nil: stage 1 of the gate has no schema yet, so only the stage shapes are
	// checked there. Stage 2 passes the real columns, which also checks that
	// every stage names an existing column.
	ClassifyTransform(stages []map[string]string, cols Columns) (reason string)
}

ArrowLane is the engine of the Arrow-native dataflow lane. It holds the type rules and the value work (cast allow-list, normalization, projection, max scan). The open build ships a stub; the official release sets newArrowLane in init().

Ownership: Normalize and Project return a record the caller owns. They do not consume the input record, so the caller releases that one itself.

func NewArrowLane added in v1.6.4 

func NewArrowLane() (ArrowLane, string)

NewArrowLane returns the lane engine, or nil and the reason it is unavailable. The closed build checks the plan token here.

type ArrowReader added in v1.4.10 

type ArrowReader struct {
	Path          string
	IpcReader     *ipc.Reader
	IpcFileReader *ipc.FileReader
	Data          *Dataset
	Context       *g.Context
	Memory        memory.Allocator
	// contains filtered or unexported fields
}

ArrowReader is a arrow reader object using arrow v18

func NewArrowFileReader added in v1.4.17 

func NewArrowFileReader(reader *os.File, selected []string) (a *ArrowReader, err error)

func NewArrowReader added in v1.4.10 

func NewArrowReader(reader io.Reader, selected []string) (a *ArrowReader, err error)

func (*ArrowReader) Columns added in v1.4.10 

func (a *ArrowReader) Columns() Columns

type ArrowWriter added in v1.4.10 

type ArrowWriter struct {
	Writer arrowRecordWriter
	// contains filtered or unexported fields
}

ArrowWriter is an arrow writer object using arrow v18

func NewArrowStreamWriter added in v1.5.18 

func NewArrowStreamWriter(w io.Writer, columns Columns, opts ...ipc.Option) (a *ArrowWriter, err error)

NewArrowStreamWriter creates an Arrow IPC stream writer (vs. the file format produced by NewArrowWriter). Use this when the destination is not seekable — e.g. stdout, a network socket, or a pipe. Stream consumers read it with pyarrow.ipc.open_stream / arrow::ipc::RecordBatchStreamReader.

func NewArrowWriter added in v1.4.10 

func NewArrowWriter(w io.Writer, columns Columns, opts ...ipc.Option) (a *ArrowWriter, err error)

NewArrowWriter creates a new Arrow IPC file writer. Extra options (e.g. ipc.WithZstd(), ipc.WithLZ4(), ipc.WithCompressConcurrency(n)) are appended after the default schema and allocator options and forwarded to ipc.NewFileWriter.

func (*ArrowWriter) Close added in v1.4.10 

func (a *ArrowWriter) Close() error

func (*ArrowWriter) Columns added in v1.4.10 

func (a *ArrowWriter) Columns() Columns

func (*ArrowWriter) WriteRecord added in v1.6.4 

func (a *ArrowWriter) WriteRecord(rec arrow.RecordBatch) error

WriteRecord writes one record to the Arrow IPC file, for the arrow lane. The record must carry the writer's schema. A record whose fields match by name and type is rebound to the writer's schema first: a reader's field metadata (driver type hints) is not part of the schema this writer stores.

func (*ArrowWriter) WriteRow added in v1.4.10 

func (a *ArrowWriter) WriteRow(row []any) error

type Avro 

type Avro struct {
	Path   string
	Reader *goavro.OCFReader
	Data   *Dataset
	// contains filtered or unexported fields
}

Avro is a avro` object

func NewAvroStream 

func NewAvroStream(reader io.ReadSeeker, columns Columns) (a *Avro, err error)

func (*Avro) Columns 

func (a *Avro) Columns() Columns

type Batch 

type Batch struct {
	Columns     Columns
	Rows        chan []any
	Previous    *Batch
	Count       int64
	Limit       int64
	MaxDuration time.Duration
	// contains filtered or unexported fields
}

func (*Batch) AddTransform 

func (b *Batch) AddTransform(transf func(row []any) []any)

func (*Batch) Close 

func (b *Batch) Close()

func (*Batch) Closed added in v1.5.23 

func (b *Batch) Closed() bool

func (*Batch) ColumnsChanged 

func (b *Batch) ColumnsChanged() bool

func (*Batch) Ds 

func (b *Batch) Ds() *Datastream

func (*Batch) ID 

func (b *Batch) ID() string

func (*Batch) IsFirst 

func (b *Batch) IsFirst() bool

func (*Batch) Push 

func (b *Batch) Push(row []any)

func (*Batch) Shape 

func (b *Batch) Shape(tgtColumns Columns, pause ...bool) (err error)

type BatchReader 

type BatchReader struct {
	Batch   *Batch
	Columns Columns
	Reader  io.Reader
	Counter int64
}

type BooleanColumnTyping added in v1.5.3 

type BooleanColumnTyping struct {
	CastAs string `json:"cast_as,omitempty" yaml:"cast_as,omitempty"` // "integer" or "string"
}

BooleanColumnTyping contains boolean type mapping configurations

func (*BooleanColumnTyping) Apply added in v1.5.3 

func (bct *BooleanColumnTyping) Apply(col *Column)

type CSV 

type CSV struct {
	Path            string
	NoHeader        bool
	SkipLines       int
	Delimiter       string
	Escape          string
	Quote           string
	FieldsPerRecord int
	Columns         []Column
	File            *os.File
	Data            Dataset
	Reader          io.Reader
	Config          map[string]string
	NoDebug         bool
	// contains filtered or unexported fields
}

CSV is a csv object

func (*CSV) InferSchema 

func (c *CSV) InferSchema() error

InferSchema returns a sample of n rows

func (*CSV) NewReader 

func (c *CSV) NewReader() (*io.PipeReader, error)

NewReader creates a Reader

func (*CSV) Read 

func (c *CSV) Read() (data Dataset, err error)

ReadStream returns the read CSV stream with Line 1 as header

func (*CSV) ReadStream 

func (c *CSV) ReadStream() (ds *Datastream, err error)

ReadStream returns the read CSV stream with Line 1 as header

func (*CSV) ReadStreamContext added in v1.1.15 

func (c *CSV) ReadStreamContext(ctx context.Context) (ds *Datastream, err error)

ReadStream returns the read CSV stream with Line 1 as header

func (*CSV) Sample 

func (c *CSV) Sample(n int) (Dataset, error)

Sample returns a sample of n rows

func (*CSV) SetFields 

func (c *CSV) SetFields(fields []string)

SetFields sets the fields

func (*CSV) WriteStream 

func (c *CSV) WriteStream(ds *Datastream) (cnt uint64, err error)

WriteStream to CSV file

type ColMetaKey added in v1.5.7 

type ColMetaKey string

ColMetaKey represents a standardized column metadata key for schema migration 

const (
	// Auto-increment/Identity
	ColMetaAutoIncrement ColMetaKey = "auto_increment"     // "true" if column is auto-increment
	ColMetaIdentitySeed  ColMetaKey = "identity_seed"      // Starting value (e.g., "1")
	ColMetaIdentityIncr  ColMetaKey = "identity_increment" // Increment step (e.g., "1")

	// Nullable
	ColMetaNullable ColMetaKey = "nullable" // "true" or "false"

	// Default Value (stores generalized form)
	ColMetaDefaultValue ColMetaKey = "default_value"

	// Primary Key
	ColMetaIsPrimaryKey ColMetaKey = "is_primary_key" // "true" if column is part of primary key

	// Unique
	ColMetaUnique ColMetaKey = "unique" // "true" if column has a UNIQUE constraint

	// Index (JSON-encoded []IndexDef contributed by this column's inline index modifiers)
	ColMetaIndex ColMetaKey = "index"

	// DDLExplicit marks that DDL constraints were declared explicitly via the
	// columns: modifier DSL (not_null, primary_key, unique, description), so they
	// should be rendered at CREATE time even when schema-migration is not enabled.
	ColMetaDDLExplicit ColMetaKey = "ddl_explicit"

	// Foreign Keys (JSON struct)
	ColMetaForeignKey ColMetaKey = "foreign_key"

	// Check Constraints (native syntax, future AI translation)
	ColMetaCheckConstraint ColMetaKey = "check_constraint"

	// Column Description/Comment
	ColMetaDescription ColMetaKey = "description"
)

func (ColMetaKey) String added in v1.5.7 

func (cmk ColMetaKey) String() string

type Column 

type Column struct {
	Position    int         `json:"position"`
	Name        string      `json:"name"`
	Type        ColumnType  `json:"type"`
	DbType      string      `json:"db_type,omitempty"`
	DbPrecision int         `json:"db_precision,omitempty"`
	DbScale     int         `json:"db_scale,omitempty"`
	Sourced     bool        `json:"-"` // whether col was sourced/inferred from a typed source
	Stats       ColumnStats `json:"stats,omitempty"`

	Table       string `json:"table,omitempty"`
	Schema      string `json:"schema,omitempty"`
	Database    string `json:"database,omitempty"`
	Description string `json:"description,omitempty"`
	FileURI     string `json:"file_uri,omitempty"`

	Constraint *ColumnConstraint `json:"constraint,omitempty"`
	Metadata   map[string]string `json:"metadata,omitempty"`
	// contains filtered or unexported fields
}

Column represents a schemata column

func InferFromStats 

func InferFromStats(columns []Column, safe bool, noDebug bool) []Column

InferFromStats using the stats to infer data types

func (*Column) EvaluateConstraint added in v1.2.16 

func (col *Column) EvaluateConstraint(value any, sp *StreamProcessor) (err error)

EvaluateConstraint evaluates a value against the constraint function

func (*Column) GetDefaultValue added in v1.5.7 

func (col *Column) GetDefaultValue() string

GetDefaultValue returns the default value metadata

func (*Column) GetForeignKey added in v1.5.7 

func (col *Column) GetForeignKey() (*ForeignKeyInfo, error)

GetForeignKey returns the foreign key info if present

func (*Column) GetIndexDefs added in v1.5.20 

func (col *Column) GetIndexDefs() (defs []IndexDef)

GetIndexDefs returns the inline index definitions declared on this column via the columns: modifier DSL (index/index(...)/unique_index(...)).

func (*Column) GetNativeType added in v1.2.16 

func (col *Column) GetNativeType(t dbio.Type, ct ColumnTyping) (nativeType string, err error)

GetNativeType returns the native column type from generic

func (*Column) GoType 

func (col *Column) GoType() reflect.Type

func (*Column) HasNulls added in v1.1.7 

func (col *Column) HasNulls() bool

func (*Column) HasNullsPlus1 added in v1.1.7 

func (col *Column) HasNullsPlus1() bool

HasNullsPlus1 denotes when a column is all nulls plus 1 non-null

func (*Column) HasUniqueConstraint added in v1.5.20 

func (col *Column) HasUniqueConstraint() bool

HasUniqueConstraint returns true if the column has a UNIQUE constraint declared via the columns: modifier DSL.

func (*Column) IsAutoIncrement added in v1.5.7 

func (col *Column) IsAutoIncrement() bool

IsAutoIncrement returns true if the column is auto-increment/identity

func (*Column) IsBinary added in v1.2.3 

func (col *Column) IsBinary() bool

IsBinary returns whether the column is a binary

func (*Column) IsBool 

func (col *Column) IsBool() bool

IsBool returns whether the column is a boolean

func (*Column) IsDDLExplicit added in v1.5.20 

func (col *Column) IsDDLExplicit() bool

IsDDLExplicit returns true if DDL constraints were declared explicitly via the columns: modifier DSL (so they apply at CREATE time without schema-migration).

func (*Column) IsDate added in v1.1.14 

func (col *Column) IsDate() bool

IsDate returns whether the column is a datet object

func (*Column) IsDatetime 

func (col *Column) IsDatetime() bool

IsDatetime returns whether the column is a datetime object

func (*Column) IsDecimal 

func (col *Column) IsDecimal() bool

IsDecimal returns whether the column is a decimal

func (*Column) IsFloat added in v1.1.14 

func (col *Column) IsFloat() bool

IsFloat returns whether the column is a float

func (*Column) IsInteger 

func (col *Column) IsInteger() bool

IsInteger returns whether the column is an integer

func (*Column) IsKeyType added in v1.1.12 

func (col *Column) IsKeyType(keyType KeyType) bool

func (*Column) IsNullable added in v1.5.7 

func (col *Column) IsNullable() bool

IsNullable returns true if the column allows NULL values Defaults to true if not explicitly set

func (*Column) IsNumber 

func (col *Column) IsNumber() bool

IsNumber returns whether the column is a decimal or an integer

func (*Column) IsPrimaryKey added in v1.5.7 

func (col *Column) IsPrimaryKey() bool

IsPrimaryKey returns true if the column is part of the primary key

func (*Column) IsString 

func (col *Column) IsString() bool

IsString returns whether the column is a string

func (*Column) IsUnique 

func (col *Column) IsUnique() bool

func (*Column) Key 

func (col *Column) Key() string

func (*Column) ParseModifiers added in v1.5.20 

func (col *Column) ParseModifiers() (err error)

ParseModifiers parses the column type slot for the modifier DSL: 

<type> [<modifier> ...]

where <modifier> is one of: not_null, nullable, primary_key, unique, description(<text>), index, index(<kwargs>), unique_index(<kwargs>).

It mutates the column's Type, Metadata, Description and inline index metadata. It must be called on the type slot only (after the `|` runtime-constraint split performed by SetConstraint).

func (*Column) SetConstraint added in v1.2.16 

func (col *Column) SetConstraint()

func (*Column) SetLengthPrecisionScale 

func (col *Column) SetLengthPrecisionScale()

SetLengthPrecisionScale parse length, precision, scale

func (*Column) SetMetadata 

func (col *Column) SetMetadata(key string, value string)

type ColumnCasing added in v1.2.22 

type ColumnCasing string

ColumnCasing is the casing method to use 

const (
	// see https://github.com/slingdata-io/sling-cli/issues/538
	NormalizeColumnCasing ColumnCasing = "normalize" // normalize to target, leaves mixed cases columns as it
	SourceColumnCasing    ColumnCasing = "source"    // keeps source column name casing. The default.
	TargetColumnCasing    ColumnCasing = "target"    // converts casing according to target database. Lower-case for files.
	SnakeColumnCasing     ColumnCasing = "snake"     // converts snake casing according to target database. Lower-case for files.
	UpperColumnCasing     ColumnCasing = "upper"     // make it upper case
	LowerColumnCasing     ColumnCasing = "lower"     // make it lower case
	CamelColumnCasing     ColumnCasing = "camel"     // converts to camelCase
)

func (*ColumnCasing) Apply added in v1.2.22 

func (cc *ColumnCasing) Apply(name string, tgtConnType dbio.Type) string

Apply applies column casing to provided name. If cc is nil or SourceColumnCasing, it returns the original value

func (*ColumnCasing) ApplyColumns added in v1.4.25 

func (cc *ColumnCasing) ApplyColumns(cols Columns, tgtType dbio.Type) (newCols Columns)

IsEmpty return true if nil or blank

func (*ColumnCasing) Equals added in v1.2.22 

func (cc *ColumnCasing) Equals(val ColumnCasing) bool

Equals evaluates equality for column casing (pointer safe)

func (*ColumnCasing) IsEmpty added in v1.2.22 

func (cc *ColumnCasing) IsEmpty() bool

IsEmpty return true if nil or blank

type ColumnConstraint added in v1.2.16 

type ColumnConstraint struct {
	Expression string             `json:"expression,omitempty"`
	Errors     []string           `json:"errors,omitempty"`
	FailCnt    uint64             `json:"fail_cnt,omitempty"`
	EvalFunc   ConstraintEvalFunc `json:"-"`
}

type ColumnStats 

type ColumnStats struct {
	MinLen       int    `json:"min_len,omitempty"`
	MaxLen       int    `json:"max_len,omitempty"`
	MaxDecLen    int    `json:"max_dec_len,omitempty"`
	Min          int64  `json:"min,omitempty"`
	Max          int64  `json:"max,omitempty"`
	NullCnt      int64  `json:"null_cnt,omitempty"`
	IntCnt       int64  `json:"int_cnt,omitempty"`
	DecCnt       int64  `json:"dec_cnt,omitempty"`
	BoolCnt      int64  `json:"bool_cnt,omitempty"`
	JsonCnt      int64  `json:"json_cnt,omitempty"`
	StringCnt    int64  `json:"string_cnt,omitempty"`
	DateCnt      int64  `json:"date_cnt,omitempty"`
	DateTimeCnt  int64  `json:"datetime_cnt,omitempty"`
	DateTimeZCnt int64  `json:"datetimez_cnt,omitempty"`
	GeometryCnt  int64  `json:"geometry_cnt,omitempty"`
	TotalCnt     int64  `json:"total_cnt,omitempty"`
	UniqCnt      int64  `json:"uniq_cnt,omitempty"`
	Checksum     uint64 `json:"checksum,omitempty"`
	LastVal      any    `json:"-"` // last non-empty value. useful for state incremental
	MaxStr       string `json:"-"` // maximum of a string update key, tracked by the arrow lane
}

ColumnStats holds statistics for a column

func (*ColumnStats) DistinctPercent 

func (cs *ColumnStats) DistinctPercent() float64

func (*ColumnStats) DuplicateCount 

func (cs *ColumnStats) DuplicateCount() int64

func (*ColumnStats) DuplicatePercent 

func (cs *ColumnStats) DuplicatePercent() float64

type ColumnType 

type ColumnType string
const (
	BigIntType     ColumnType = "bigint"
	BinaryType     ColumnType = "binary"
	BoolType       ColumnType = "bool"
	DateType       ColumnType = "date"
	DatetimeType   ColumnType = "datetime"
	DecimalType    ColumnType = "decimal"
	IntegerType    ColumnType = "integer"
	JsonType       ColumnType = "json"
	SmallIntType   ColumnType = "smallint"
	StringType     ColumnType = "string"
	UUIDType       ColumnType = "uuid"
	TextType       ColumnType = "text"
	TimestampType  ColumnType = "timestamp"
	TimestampzType ColumnType = "timestampz"
	FloatType      ColumnType = "float"
	TimeType       ColumnType = "time"
	TimezType      ColumnType = "timez"
	GeometryType   ColumnType = "geometry"
)

func NativeTypeToGeneral added in v1.2.19 

func NativeTypeToGeneral(name, dbType string, connType dbio.Type) (colType ColumnType)

func (ColumnType) IsBinary added in v1.2.3 

func (ct ColumnType) IsBinary() bool

IsBinary returns whether the column is a binary

func (ColumnType) IsBool 

func (ct ColumnType) IsBool() bool

IsBool returns whether the column is a boolean

func (ColumnType) IsDate added in v1.1.8 

func (ct ColumnType) IsDate() bool

IsDatetime returns whether the column is a datetime object

func (ColumnType) IsDatetime 

func (ct ColumnType) IsDatetime() bool

IsDatetime returns whether the column is a datetime object

func (ColumnType) IsDecimal 

func (ct ColumnType) IsDecimal() bool

IsDecimal returns whether the column is a decimal

func (ColumnType) IsFloat added in v1.1.14 

func (ct ColumnType) IsFloat() bool

IsFloat returns whether the column is a float

func (ColumnType) IsGeometry added in v1.5.2 

func (ct ColumnType) IsGeometry() bool

IsGeometry returns whether the column is a geometry object

func (ColumnType) IsInteger 

func (ct ColumnType) IsInteger() bool

IsInteger returns whether the column is an integer

func (ColumnType) IsJSON 

func (ct ColumnType) IsJSON() bool

IsJSON returns whether the column is a json

func (ColumnType) IsNumber 

func (ct ColumnType) IsNumber() bool

IsNumber returns whether the column is a decimal or an integer

func (ColumnType) IsString 

func (ct ColumnType) IsString() bool

IsString returns whether the column is a string

func (ColumnType) IsTime added in v1.4.17 

func (ct ColumnType) IsTime() bool

IsTime returns whether the column is a time object

func (ColumnType) IsValid 

func (ct ColumnType) IsValid() bool

IsValid returns whether the column has a valid type

type ColumnTyping added in v1.4.5 

type ColumnTyping struct {
	String  *StringColumnTyping  `json:"string,omitempty" yaml:"string,omitempty"`
	Decimal *DecimalColumnTyping `json:"decimal,omitempty" yaml:"decimal,omitempty"`
	JSON    *JsonColumnTyping    `json:"json,omitempty" yaml:"json,omitempty"`
	Boolean *BooleanColumnTyping `json:"boolean,omitempty" yaml:"boolean,omitempty"`
	UUID    *UUIDColumnTyping    `json:"uuid,omitempty" yaml:"uuid,omitempty"`
}

ColumnTyping contains type-specific mapping configurations

func (*ColumnTyping) MaxDecimals added in v1.4.19 

func (ct *ColumnTyping) MaxDecimals() int

type Columns 

type Columns []Column

Columns represent many columns

func ArrowSchemaToColumns added in v1.4.10 

func ArrowSchemaToColumns(schema *arrow.Schema) Columns

ArrowSchemaToColumns converts arrow schema to Columns

func NewColumns 

func NewColumns(cols ...Column) Columns

NewColumns normalizes Columns: defaults missing/invalid Type to StringType and assigns 1-based Position.

func NewColumnsFromFields 

func NewColumnsFromFields(fields ...string) (cols Columns)

NewColumnsFromFields creates Columns from fields

func (Columns) Clone 

func (cols Columns) Clone() (newCols Columns)

Names return the column names

func (Columns) Coerce 

func (cols Columns) Coerce(castCols Columns, hasHeader bool, casing ColumnCasing, tgtType dbio.Type) (newCols Columns)

Coerce casts columns into specified types

func (Columns) CollectInlineIndexes added in v1.5.20 

func (cols Columns) CollectInlineIndexes(tableName string, maxLength int) (defs []IndexDef, err error)

CollectInlineIndexes gathers the inline index definitions declared across all columns via the columns: modifier DSL and merges them into final []IndexDef.

Columns sharing the same index Name form a composite index, ordered by Priority. Per-column attributes are Priority and Sort; shared attributes (Where, Type, Unique, Include) must be identical (or omitted) across every column declaring the same named index, else an error is returned. Anonymous indexes (no Name) are auto-named idx_<table>_<col> and never merged.

func (Columns) Data added in v1.2.2 

func (cols Columns) Data(includeParent bool) (fields []string, rows [][]any)

func (Columns) Dataset 

func (cols Columns) Dataset() Dataset

Dataset return an empty inferred dataset

func (Columns) DbTypes 

func (cols Columns) DbTypes(args ...bool) []string

DbTypes return the column names/db types args -> (lower bool, cleanUp bool)

func (Columns) FieldMap 

func (cols Columns) FieldMap(toLower bool) map[string]int

FieldMap return the fields map of indexes when `toLower` is true, field keys are lower cased

func (Columns) GetColumn 

func (cols Columns) GetColumn(name string) *Column

GetColumn returns the matched Col

func (Columns) GetKeys 

func (cols Columns) GetKeys(keyType KeyType) Columns

GetKeys gets key columns

func (Columns) GetMissing added in v1.1.8 

func (cols Columns) GetMissing(newCols ...Column) (missing Columns)

GetMissing returns the missing columns from newCols

func (Columns) IsDifferent 

func (cols Columns) IsDifferent(newCols Columns) bool

func (Columns) IsDummy 

func (cols Columns) IsDummy() bool

IsDummy returns true if the columns are injected by CreateDummyFields

func (Columns) IsSimilarTo 

func (cols Columns) IsSimilarTo(otherCols Columns) bool

IsSimilarTo returns true if has same number of columns and contains the same columns, but may be in different order

func (Columns) JSON added in v1.2.2 

func (cols Columns) JSON(includeParent bool) (output string)

PrettyTable returns a text pretty table

func (Columns) KeepSourcedTypes added in v1.6.4 

func (cols Columns) KeepSourcedTypes(described Columns) Columns

KeepSourcedTypes sets the types of described on cols, when the two have the same column names. A described type (e.g. json, uuid, decimal precision) is more exact than the type of an Arrow schema.

func (Columns) Keys 

func (cols Columns) Keys() []string

Names return the column names args -> (lower bool, cleanUp bool)

func (Columns) MakeRec 

func (cols Columns) MakeRec(row []any) map[string]any

func (Columns) MakeShaper 

func (cols Columns) MakeShaper(tgtColumns Columns) (shaper *Shaper, err error)

func (Columns) Map added in v1.5.1 

func (cols Columns) Map() map[string]*Column

Map return the map of columns

func (Columns) Merge added in v1.1.15 

func (cols Columns) Merge(newCols Columns, overwrite bool) (col2 Columns, added schemaChg, changed []schemaChg)

func (Columns) Names 

func (cols Columns) Names(args ...bool) []string

Names return the column names args -> (lower bool, cleanUp bool)

func (Columns) PrettyTable added in v1.1.8 

func (cols Columns) PrettyTable(includeParent bool) (output string)

PrettyTable returns a text pretty table

func (Columns) PrimaryKeyNames added in v1.6.4 

func (cols Columns) PrimaryKeyNames() (names []string)

PrimaryKeyNames returns the names of the columns with the primary key constraint (from schema migration metadata or the columns DSL)

func (Columns) SetKeys 

func (cols Columns) SetKeys(keyType KeyType, colNames ...string) (err error)

SetKeys sets key columns

func (Columns) SetMetadata added in v1.2.15 

func (cols Columns) SetMetadata(key, value string, colNames ...string) (err error)

SetMetadata sets metadata for columns

func (Columns) Sourced added in v1.1.6 

func (cols Columns) Sourced() (sourced bool)

Sourced returns true if the columns are all sourced

func (Columns) Types 

func (cols Columns) Types(args ...bool) []string

Types return the column names/types args -> (lower bool, cleanUp bool)

func (Columns) ValidateNames added in v1.4.5 

func (cols Columns) ValidateNames(tgtType dbio.Type) (newCols Columns)

ValidateNames truncates the column name it exceed the max column length

func (Columns) WithoutMeta added in v1.2.2 

func (cols Columns) WithoutMeta() (newCols Columns)

WithoutMeta returns the columns with metadata columns

type Compressor 

type Compressor interface {
	Self() Compressor
	Compress(io.Reader) io.Reader
	Decompress(io.Reader) (io.Reader, error)
	Suffix() string
}

Compressor implements differnt kind of compression

func NewCompressor 

func NewCompressor(cpType CompressorType) Compressor

type CompressorType 

type CompressorType string

CompressorType is an int type for enum for the Compressor Type 

const (
	// AutoCompressorType is for auto compression
	AutoCompressorType CompressorType = "auto"
	// NoneCompressorType is for no compression
	NoneCompressorType CompressorType = "none"
	// ZipCompressorType is for Zip compression
	ZipCompressorType CompressorType = "zip"
	// GzipCompressorType is for Gzip compression
	GzipCompressorType CompressorType = "gzip"
	// SnappyCompressorType is for Snappy compression
	SnappyCompressorType CompressorType = "snappy"
	// ZStandardCompressorType is for ZStandard
	ZStandardCompressorType CompressorType = "zstd"
)

func CompressorTypePtr 

func CompressorTypePtr(v CompressorType) *CompressorType

CompressorTypePtr returns a pointer to the CompressorType value passed in.

func (CompressorType) Normalize added in v1.3.4 

func (ct CompressorType) Normalize() *CompressorType

Normalize converts to lowercase

func (CompressorType) String added in v1.3.4 

func (ct CompressorType) String() string

String converts to lowercase

type ConstraintEvalFunc added in v1.2.16 

type ConstraintEvalFunc func(value any) bool

type CsvDuckDb added in v1.2.19 

type CsvDuckDb struct {
	URI  string
	Duck *DuckDb
	// contains filtered or unexported fields
}

func NewCsvReaderDuckDb added in v1.2.19 

func NewCsvReaderDuckDb(uri string, sc *StreamConfig, props ...string) (*CsvDuckDb, error)

func (*CsvDuckDb) Close added in v1.2.19 

func (r *CsvDuckDb) Close() error

func (*CsvDuckDb) Columns added in v1.2.19 

func (r *CsvDuckDb) Columns() (Columns, error)

func (*CsvDuckDb) MakeQuery added in v1.2.19 

func (r *CsvDuckDb) MakeQuery(fsc FileStreamConfig) string

type Dataflow 

type Dataflow struct {
	Columns     Columns
	Buffer      [][]interface{}
	StreamCh    chan *Datastream
	Streams     []*Datastream
	Context     *g.Context
	Limit       uint64
	EgressBytes uint64

	Ready           bool
	Inferred        bool
	FsURL           string
	OnColumnChanged func(col Column) error
	OnColumnAdded   func(col Column) error

	StreamMap map[string]*Datastream

	SchemaVersion int // for column type version
	// contains filtered or unexported fields
}

Dataflow is a collection of concurrent Datastreams

func MakeDataFlow 

func MakeDataFlow(dss ...*Datastream) (df *Dataflow, err error)

MakeDataFlow create a dataflow from datastreams

func NewDataflow 

func NewDataflow(limit ...int) (df *Dataflow)

NewDataflow creates a new dataflow

func NewDataflowContext added in v1.1.15 

func NewDataflowContext(ctx context.Context, limit ...int) (df *Dataflow)

func (*Dataflow) AddColumns 

func (df *Dataflow) AddColumns(newCols Columns, overwrite bool, exceptDs ...string) (added Columns, processOk bool)

SetColumns sets the columns

func (*Dataflow) AddEgressBytes added in v1.2.2 

func (df *Dataflow) AddEgressBytes(bytes uint64)

AddEgressBytes add egress bytes

func (*Dataflow) ArrowOnly added in v1.6.4 

func (df *Dataflow) ArrowOnly() bool

ArrowOnly returns true when every stream of the dataflow is an Arrow stream. A dataflow with no streams is not ArrowOnly.

func (*Dataflow) BufferDataset added in v1.2.25 

func (df *Dataflow) BufferDataset() Dataset

BufferDataset return the buffer as a dataset

func (*Dataflow) Bytes 

func (df *Dataflow) Bytes() (inBytes, outBytes uint64)

func (*Dataflow) ChangeColumn 

func (df *Dataflow) ChangeColumn(i int, newType ColumnType, exceptDs ...string) bool

SetColumns sets the columns

func (*Dataflow) CleanUp 

func (df *Dataflow) CleanUp()

CleanUp refers the defer functions

func (*Dataflow) Close 

func (df *Dataflow) Close()

Close closes the df

func (*Dataflow) CloseCurrentBatches 

func (df *Dataflow) CloseCurrentBatches()

func (*Dataflow) Collect 

func (df *Dataflow) Collect() (data Dataset, err error)

Collect reads from one or more streams and return a dataset

func (*Dataflow) Count 

func (df *Dataflow) Count() (cnt uint64)

Count returns the aggregate count

func (*Dataflow) Defer 

func (df *Dataflow) Defer(f func())

Defer runs a given function as close of Dataflow

func (*Dataflow) DsTotalBytes 

func (df *Dataflow) DsTotalBytes() (bytes uint64)

func (*Dataflow) Err 

func (df *Dataflow) Err() (err error)

Err return the error if any

func (*Dataflow) IsClosed 

func (df *Dataflow) IsClosed() bool

IsClosed is true is ds is closed

func (*Dataflow) IsEmpty 

func (df *Dataflow) IsEmpty() bool

IsEmpty returns true is ds.Rows of all channels as empty

func (*Dataflow) MakeStreamCh 

func (df *Dataflow) MakeStreamCh(forceMerge bool) (streamCh chan *Datastream)

MakeStreamCh determines whether to merge all the streams into one or keep them separate. If data is small per stream, it's best to merge For example, Bigquery has limits on number of operations can be called within a time limit

func (*Dataflow) MergeColumns added in v1.1.15 

func (df *Dataflow) MergeColumns(columns []Column, inferred bool) (processOk bool)

SetColumns sets the columns

func (*Dataflow) Pause 

func (df *Dataflow) Pause(exceptDs ...string) bool

Pause pauses all streams. Returns false if any stream could not be paused within the pause timeout or before the dataflow context is canceled.

func (*Dataflow) PropagateColum added in v1.4.24 

func (df *Dataflow) PropagateColum(colIndex int)

PropagateColum propagates the dataflow column properties to underlying datastreams

func (*Dataflow) PushStreamChan 

func (df *Dataflow) PushStreamChan(dsCh chan *Datastream)

func (*Dataflow) SetBatchLimit added in v1.2.11 

func (df *Dataflow) SetBatchLimit(limit int64)

SetBatchLimit set the ds.Batch.Limit

func (*Dataflow) SetBatchMaxDuration added in v1.5.23 

func (df *Dataflow) SetBatchMaxDuration(d time.Duration)

SetBatchMaxDuration sets a wall-clock ceiling for each batch (commit cadence).

func (*Dataflow) SetConfig added in v1.1.15 

func (df *Dataflow) SetConfig(cfg StreamConfig)

SetConfig set the Sp config

func (*Dataflow) SetEmpty 

func (df *Dataflow) SetEmpty()

SetEmpty sets all underlying datastreams empty

func (*Dataflow) SetReady 

func (df *Dataflow) SetReady()

SetReady sets the df.ready

func (*Dataflow) Size 

func (df *Dataflow) Size() int

Size is the number of streams

func (*Dataflow) StreamConfig added in v1.2.15 

func (df *Dataflow) StreamConfig() (cfg StreamConfig)

StreamConfig get the first Sp config

func (*Dataflow) SyncColumns 

func (df *Dataflow) SyncColumns()

SyncColumns a workaround to synch the ds.Columns to the df.Columns

func (*Dataflow) SyncStats 

func (df *Dataflow) SyncStats()

SyncStats sync stream processor stats aggregated to the df.Columns

func (*Dataflow) SyncedSeqValue added in v1.5.10 

func (df *Dataflow) SyncedSeqValue() (value int64)

func (*Dataflow) Unpause 

func (df *Dataflow) Unpause(exceptDs ...string)

Unpause unpauses all streams

func (*Dataflow) WaitClosed 

func (df *Dataflow) WaitClosed()

WaitClosed waits until dataflow is closed hack to make sure all streams are pushed

func (*Dataflow) WaitReady 

func (df *Dataflow) WaitReady() error

WaitReady waits until dataflow is ready

type Dataset 

type Dataset struct {
	Result        *sqlx.Rows       `json:"-"`
	Columns       Columns          `json:"columns"`
	Rows          [][]any          `json:"rows"`
	SQL           string           `json:"sql"`
	Duration      float64          `json:"duration"`
	Sp            *StreamProcessor `json:"-"`
	Inferred      bool             `json:"inferred"`
	SafeInference bool             `json:"safe_inference"`
	NoDebug       bool             `json:"no_debug"`
}

Dataset is a query returned dataset

func NewDataset 

func NewDataset(columns Columns) (data Dataset)

NewDataset return a new dataset

func NewDatasetFromMap 

func NewDatasetFromMap(m map[string]any) (data Dataset)

NewDatasetFromMap return a new dataset

func NewExcelDataset added in v1.2.2 

func NewExcelDataset(reader io.Reader, props map[string]string) (data Dataset, err error)

func ReadCsv 

func ReadCsv(path string) (Dataset, error)

ReadCsv reads CSV and returns dataset

func (*Dataset) AddColumns 

func (data *Dataset) AddColumns(newCols Columns, overwrite bool) (added Columns)

SetColumns sets the columns

func (*Dataset) Append 

func (data *Dataset) Append(row ...[]any)

Append appends a new row

func (*Dataset) ColValues 

func (data *Dataset) ColValues(col int) []any

ColValues returns the values of a one column as array

func (*Dataset) ColValuesStr 

func (data *Dataset) ColValuesStr(col int) []string

ColValuesStr returns the values of a one column as array or string

func (*Dataset) FirstRow 

func (data *Dataset) FirstRow() []any

FirstRow returns the first row

func (*Dataset) FirstVal 

func (data *Dataset) FirstVal() any

FirstVal returns the first value from the first row

func (*Dataset) GetFields 

func (data *Dataset) GetFields(lower ...bool) []string

GetFields return the fields of the Data

func (*Dataset) InferColumnTypes 

func (data *Dataset) InferColumnTypes()

InferColumnTypes determines the columns types

func (*Dataset) Pick 

func (data *Dataset) Pick(colNames ...string) (nData Dataset)

Pick returns a new dataset with specified columns

func (*Dataset) PrettyTable added in v1.1.8 

func (data *Dataset) PrettyTable(fields ...string) (output string)

func (*Dataset) Print 

func (data *Dataset) Print(limit int)

Print pretty prints the data with a limit 0 is unlimited

func (*Dataset) Records 

func (data *Dataset) Records(lower ...bool) []map[string]any

Records return rows of maps

func (*Dataset) RecordsCasted 

func (data *Dataset) RecordsCasted(lower ...bool) []map[string]any

RecordsCasted return rows of maps or casted values

func (*Dataset) RecordsString 

func (data *Dataset) RecordsString(lower ...bool) []map[string]string

RecordsString return rows of maps or string values

func (*Dataset) SetFields 

func (data *Dataset) SetFields(fields []string)

SetFields sets the fields/columns of the Datastream

func (*Dataset) Sort 

func (data *Dataset) Sort(args ...any)

Sort sorts by cols example: `data.Sort(0, 2, 3, false)` will sort col0, col2, col3 descending example: `data.Sort(0, 2, true)` will sort col0, col2 ascending

func (*Dataset) Stream 

func (data *Dataset) Stream(Props ...map[string]string) *Datastream

Stream returns a datastream of the dataset

func (*Dataset) ToJSONMap 

func (data *Dataset) ToJSONMap() map[string]any

ToJSONMap converst to a JSON object

func (*Dataset) WriteCsv 

func (data *Dataset) WriteCsv(dest io.Writer) (tbw int, err error)

WriteCsv writes to a writer

type Datastream 

type Datastream struct {
	Columns       Columns
	Buffer        [][]any
	BatchChan     chan *Batch
	Batches       []*Batch
	CurrentBatch  *Batch
	Count         uint64
	Context       *g.Context
	Ready         bool
	Bytes         atomic.Uint64
	Sp            *StreamProcessor
	SafeInference bool
	SchemaOnly    bool
	NoDebug       bool
	Inferred      bool

	ID       string
	Metadata Metadata // map of column name to metadata type

	ArrowOnly bool
	// contains filtered or unexported fields
}

Datastream is a stream of rows

func MergeDataflow 

func MergeDataflow(df *Dataflow) (dsN *Datastream)

MergeDataflow merges the dataflow streams into one

func NewDatastream 

func NewDatastream(columns Columns) (ds *Datastream)

NewDatastream return a new datastream

func NewDatastreamArrow added in v1.6.4 

func NewDatastreamArrow(ctx context.Context, columns Columns, rs *RecordStream) *Datastream

NewDatastreamArrow returns a datastream that carries arrow records instead of rows. The sink pulls the records from the RecordStream; the row machinery (Batch, bwRows, the iterator loop) never starts.

The caller must set ds.Columns to the Sling columns derived from rs.Schema before Start, or pass them here.

func NewDatastreamContext 

func NewDatastreamContext(ctx context.Context, columns Columns) (ds *Datastream)

NewDatastreamContext return a new datastream

func NewDatastreamIt 

func NewDatastreamIt(ctx context.Context, columns Columns, nextFunc func(it *Iterator) bool) (ds *Datastream)

NewDatastreamIt with it

func ReadCsvStream 

func ReadCsvStream(path string) (ds *Datastream, err error)

ReadCsvStream reads CSV and returns datasream

func (*Datastream) AddBytes 

func (ds *Datastream) AddBytes(b int64)

AddBytes add bytes as processed

func (*Datastream) AddColumns 

func (ds *Datastream) AddColumns(newCols Columns, overwrite bool) (added Columns)

SetColumns sets the columns

func (*Datastream) CastRowToString 

func (ds *Datastream) CastRowToString(row []any) []string

CastRowToString returns the row as string casted

func (*Datastream) CastToStringSafeMask added in v1.2.14 

func (ds *Datastream) CastToStringSafeMask(row []any) []string

CastToStringSafeMask returns the row as string mask casted ( evensafer)

func (*Datastream) ChangeColumn 

func (ds *Datastream) ChangeColumn(i int, newType ColumnType)

ChangeColumn applies a column type change

func (*Datastream) Chunk 

func (ds *Datastream) Chunk(limit uint64) (chDs chan *Datastream)

Chunk splits the datastream into chunk datastreams (in sequence)

func (*Datastream) Close 

func (ds *Datastream) Close()

Close closes the datastream

func (*Datastream) Collect 

func (ds *Datastream) Collect(limit int) (Dataset, error)

Collect reads a stream and return a dataset limit of 0 is unlimited

func (*Datastream) ConsumeArrowReader added in v1.4.10 

func (ds *Datastream) ConsumeArrowReader(reader io.Reader) (err error)

func (*Datastream) ConsumeArrowReaderSeeker added in v1.4.10 

func (ds *Datastream) ConsumeArrowReaderSeeker(reader *os.File) (err error)

ConsumeArrowReaderSeeker uses the provided reader to stream rows

func (*Datastream) ConsumeArrowReaderStream added in v1.4.17 

func (ds *Datastream) ConsumeArrowReaderStream(reader io.Reader) (err error)

ConsumeArrowReaderStream uses the provided reader to stream rows

func (*Datastream) ConsumeArrowRecords added in v1.6.4 

func (ds *Datastream) ConsumeArrowRecords(file *os.File, lane ArrowLane) (err error)

ConsumeArrowRecords streams an Arrow IPC file as records, for the arrow lane. The file schema is the stream schema: the cache writer built it with ColumnsToArrowSchema, so it already carries the Sling-derived types. The file is closed when the datastream closes.

func (*Datastream) ConsumeAvroReader 

func (ds *Datastream) ConsumeAvroReader(reader io.Reader) (err error)

ConsumeAvroReader uses the provided reader to stream rows

func (*Datastream) ConsumeAvroReaderSeeker 

func (ds *Datastream) ConsumeAvroReaderSeeker(reader io.ReadSeeker) (err error)

ConsumeAvroReaderSeeker uses the provided reader to stream rows

func (*Datastream) ConsumeCsvReader 

func (ds *Datastream) ConsumeCsvReader(reader io.Reader) (err error)

ConsumeCsvReader uses the provided reader to stream rows

func (*Datastream) ConsumeCsvReaderChl added in v1.2.4 

func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan *ReaderReady) (err error)

ConsumeCsvReaderChl reads a channel of readers. Should be safe to use with header top row

func (*Datastream) ConsumeCsvReaderDuckDb added in v1.2.19 

func (ds *Datastream) ConsumeCsvReaderDuckDb(uri string, sc FileStreamConfig) (err error)

ConsumeCsvReaderDuckDb uses the provided reader to stream rows

func (*Datastream) ConsumeDeltaReader added in v1.2.16 

func (ds *Datastream) ConsumeDeltaReader(uri string, sc FileStreamConfig) (err error)

ConsumeDeltaReader uses the provided reader to stream rows

func (*Datastream) ConsumeExcelReader added in v1.2.2 

func (ds *Datastream) ConsumeExcelReader(reader io.Reader, props map[string]string) (err error)

ConsumeSASReader uses the provided reader to stream rows

func (*Datastream) ConsumeExcelReaderSeeker added in v1.2.2 

func (ds *Datastream) ConsumeExcelReaderSeeker(reader io.ReadSeeker, props map[string]string) (err error)

ConsumeSASReaderSeeker uses the provided reader to stream rows

func (*Datastream) ConsumeIcebergReader added in v1.2.16 

func (ds *Datastream) ConsumeIcebergReader(uri string, sc FileStreamConfig) (err error)

ConsumeIcebergReader uses the provided reader to stream rows

func (*Datastream) ConsumeJsonReader 

func (ds *Datastream) ConsumeJsonReader(reader io.Reader) (err error)

ConsumeJsonReader uses the provided reader to stream JSON This will put each JSON rec as one string value so payload can be processed downstream

func (*Datastream) ConsumeJsonReaderChl added in v1.2.6 

func (ds *Datastream) ConsumeJsonReaderChl(readerChn chan *ReaderReady, isXML bool) (err error)

func (*Datastream) ConsumeParquetReader 

func (ds *Datastream) ConsumeParquetReader(reader io.Reader) (err error)

ConsumeParquetReader uses the provided reader to stream rows

func (*Datastream) ConsumeParquetReaderDuckDb added in v1.2.16 

func (ds *Datastream) ConsumeParquetReaderDuckDb(uri string, sc FileStreamConfig) (err error)

ConsumeParquetReader uses the provided reader to stream rows

func (*Datastream) ConsumeParquetReaderSeeker 

func (ds *Datastream) ConsumeParquetReaderSeeker(reader *os.File) (err error)

ConsumeParquetReader uses the provided reader to stream rows

func (*Datastream) ConsumeSASReader 

func (ds *Datastream) ConsumeSASReader(reader io.Reader) (err error)

ConsumeSASReader uses the provided reader to stream rows

func (*Datastream) ConsumeSASReaderSeeker 

func (ds *Datastream) ConsumeSASReaderSeeker(reader io.ReadSeeker) (err error)

ConsumeSASReaderSeeker uses the provided reader to stream rows

func (*Datastream) ConsumeXmlReader 

func (ds *Datastream) ConsumeXmlReader(reader io.Reader) (err error)

ConsumeXmlReader uses the provided reader to stream XML This will put each XML rec as one string value so payload can be processed downstream

func (*Datastream) Defer 

func (ds *Datastream) Defer(f func())

Defer runs a given function as close of Datastream

func (*Datastream) Df 

func (ds *Datastream) Df() *Dataflow

func (*Datastream) DrainReprocess added in v1.5.9 

func (ds *Datastream) DrainReprocess() []any

DrainReprocess returns any row that was sent to the Reprocess channel during a pause/unpause cycle but was not consumed (e.g., for push-based streams that have no ds.Start() iterator loop). Returns nil if empty.

func (*Datastream) Err 

func (ds *Datastream) Err() (err error)

Err return the error if any

func (*Datastream) GetConfig 

func (ds *Datastream) GetConfig() (configMap map[string]string)

GetConfig get config

func (*Datastream) GetFields 

func (ds *Datastream) GetFields(args ...bool) []string

GetFields return the fields of the Data

func (*Datastream) IsClosed 

func (ds *Datastream) IsClosed() bool

IsClosed is true is ds is closed

func (*Datastream) LatestBatch 

func (ds *Datastream) LatestBatch() *Batch

func (*Datastream) Limited added in v1.2.4 

func (ds *Datastream) Limited(limit ...int) bool

func (*Datastream) Map 

func (ds *Datastream) Map(newColumns Columns, transf func([]any) []any) (nDs *Datastream)

Map applies the provided function to every row and returns the result

func (*Datastream) MapParallel 

func (ds *Datastream) MapParallel(transf func([]any) []any, numWorkers int) (nDs *Datastream)

MapParallel applies the provided function to every row in parallel and returns the result. Order is not maintained.

func (*Datastream) NewArrowReaderChnl added in v1.4.10 

func (ds *Datastream) NewArrowReaderChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewArrowReaderChnl provides a channel of readers as the limit is reached each channel flows as fast as the consumer consumes

func (*Datastream) NewArrowRecordChnl added in v1.6.4 

func (ds *Datastream) NewArrowRecordChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewArrowRecordChnl provides a channel of Arrow IPC readers built straight from the records of an Arrow stream. Same shape as NewParquetRecordChnl.

func (*Datastream) NewBatch 

func (ds *Datastream) NewBatch(columns Columns) *Batch

NewBatch create new batch with fixed columns should be used each time column type changes, or columns are added

func (*Datastream) NewCsvBufferReader 

func (ds *Datastream) NewCsvBufferReader(sc StreamConfig) *bytes.Reader

NewCsvBufferReader creates a Reader with limit. If limit == 0, then read all rows.

func (*Datastream) NewCsvBufferReaderChnl 

func (ds *Datastream) NewCsvBufferReaderChnl(sc StreamConfig) (readerChn chan *bytes.Reader)

NewCsvBufferReaderChnl provides a channel of readers as the limit is reached data is read in memory, whereas NewCsvReaderChnl does not hold in memory

func (*Datastream) NewCsvBytesChnl 

func (ds *Datastream) NewCsvBytesChnl(sc StreamConfig) (dataChn chan *[]byte)

NewCsvBytesChnl returns a channel yield chunk of bytes of csv

func (*Datastream) NewCsvReader 

func (ds *Datastream) NewCsvReader(sc StreamConfig) *io.PipeReader

NewCsvReader creates a Reader with limit. If limit == 0, then read all rows.

func (*Datastream) NewCsvReaderChnl 

func (ds *Datastream) NewCsvReaderChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewCsvReaderChnl provides a channel of readers as the limit is reached each channel flows as fast as the consumer consumes

func (*Datastream) NewExcelReaderChnl added in v1.2.2 

func (ds *Datastream) NewExcelReaderChnl(sc StreamConfig) (readerChn chan *BatchReader)

func (*Datastream) NewGeojsonReaderChnl added in v1.5.2 

func (ds *Datastream) NewGeojsonReaderChnl(sc StreamConfig) (readerChn chan *io.PipeReader)

NewGeojsonReaderChnl creates a GeoJSON FeatureCollection reader from datastream rows. Requires a column with type "geometry" to be specified in the columns section. If not provided, it will fallback to "geometry". The geometry column becomes the Feature geometry, all other columns become Feature properties.

func (*Datastream) NewIterator 

func (ds *Datastream) NewIterator(columns Columns, nextFunc func(it *Iterator) bool) *Iterator

func (*Datastream) NewJsonLinesReaderChnl 

func (ds *Datastream) NewJsonLinesReaderChnl(sc StreamConfig) (readerChn chan *io.PipeReader)

NewJsonLinesReaderChnl provides a channel of readers as the limit is reached each channel flows as fast as the consumer consumes

func (*Datastream) NewJsonReaderChnl 

func (ds *Datastream) NewJsonReaderChnl(sc StreamConfig) (readerChn chan *io.PipeReader)

func (*Datastream) NewParquetArrowReaderChnl added in v1.1.7 

func (ds *Datastream) NewParquetArrowReaderChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewParquetArrowReaderChnl provides a channel of readers as the limit is reached each channel flows as fast as the consumer consumes WARN: Not using this one since it doesn't write Decimals properly.

func (*Datastream) NewParquetReaderChnl 

func (ds *Datastream) NewParquetReaderChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewParquetReaderChnl provides a channel of readers as the limit is reached each channel flows as fast as the consumer consumes

func (*Datastream) NewParquetRecordChnl added in v1.6.4 

func (ds *Datastream) NewParquetRecordChnl(sc StreamConfig) (readerChn chan *BatchReader)

NewParquetRecordChnl provides a channel of Parquet readers built straight from the records of an Arrow stream. No row is built and no builder is refilled: the record's buffers go to the writer as they are.

FileMaxRows splits a record with a zero-copy slice, so part row counts are exact. FileMaxBytes is approximate: the writer rolls over at the first record boundary after the written bytes pass the limit, since the final compressed size is only known after the write.

func (*Datastream) Pause 

func (ds *Datastream) Pause()

func (*Datastream) Push 

func (ds *Datastream) Push(row []any)

Push return the fields of the Data

func (*Datastream) RecordStream added in v1.6.4 

func (ds *Datastream) RecordStream() *RecordStream

RecordStream returns the arrow stream of an Arrow datastream, nil otherwise.

func (*Datastream) Records 

func (ds *Datastream) Records() <-chan map[string]any

Records return rows of maps

func (*Datastream) Rows 

func (ds *Datastream) Rows() chan []any

func (*Datastream) SetConfig 

func (ds *Datastream) SetConfig(configMap map[string]string)

SetConfig sets the ds.config values

func (*Datastream) SetEmpty 

func (ds *Datastream) SetEmpty()

SetEmpty sets the ds.Rows channel as empty

func (*Datastream) SetFields 

func (ds *Datastream) SetFields(fields []string)

SetFields sets the fields/columns of the Datastream

func (*Datastream) SetFileURI added in v1.1.15 

func (ds *Datastream) SetFileURI()

SetFileURI sets the FileURI of the columns of the Datastream

func (*Datastream) SetIterator added in v1.1.14 

func (ds *Datastream) SetIterator(it *Iterator)

func (*Datastream) SetMetadata 

func (ds *Datastream) SetMetadata(jsonStr string)

func (*Datastream) SetReady 

func (ds *Datastream) SetReady()

SetReady sets the ds.ready

func (*Datastream) Shape 

func (ds *Datastream) Shape(columns Columns) (nDs *Datastream, err error)

Shape changes the column types as needed, to the provided columns var It will cast the already wrongly casted rows, and not recast the correctly casted rows

func (*Datastream) Split 

func (ds *Datastream) Split(numStreams ...int) (dss []*Datastream)

Split splits the datastream into parallel datastreams

func (*Datastream) Start 

func (ds *Datastream) Start() (err error)

func (*Datastream) StartBwProcessor added in v1.5.5 

func (ds *Datastream) StartBwProcessor()

StartBwProcessor starts the bytes-written processor goroutine. This should be called when creating a datastream that pushes rows directly without calling Start() (e.g., chunked streaming).

func (*Datastream) TryPause 

func (ds *Datastream) TryPause() bool

func (*Datastream) Unpause 

func (ds *Datastream) Unpause()

Unpause unpauses all streams

func (*Datastream) WaitClosed 

func (ds *Datastream) WaitClosed()

WaitClosed waits until dataflow is closed hack to make sure all streams are pushed

func (*Datastream) WaitReady 

func (ds *Datastream) WaitReady() error

WaitReady waits until datastream is ready

type DecimalColumnTyping added in v1.4.5 

type DecimalColumnTyping struct {
	MinPrecision *int   `json:"min_precision,omitempty" yaml:"min_precision,omitempty"` // Total number of digits
	MaxPrecision int    `json:"max_precision,omitempty" yaml:"max_precision,omitempty"` // Total number of digits
	MinScale     *int   `json:"min_scale,omitempty" yaml:"min_scale,omitempty"`         // Number of digits after decimal point
	MaxScale     int    `json:"max_scale,omitempty" yaml:"max_scale,omitempty"`
	CastAs       string `json:"cast_as,omitempty" yaml:"cast_as,omitempty"`
}

DecimalColumnTyping contains decimal type mapping configurations

func (*DecimalColumnTyping) Apply added in v1.4.5 

func (dct *DecimalColumnTyping) Apply(col Column) (precision, scale int)

type DefaultExpr added in v1.5.7 

type DefaultExpr string

DefaultExpr represents a generalized default expression These are database-agnostic representations translated to native syntax 

const (
	// Timestamp/DateTime defaults
	DefaultExprCurrentTimestamp    DefaultExpr = "current_timestamp"
	DefaultExprCurrentTimestampUTC DefaultExpr = "current_timestamp_utc"
	DefaultExprCurrentDate         DefaultExpr = "current_date"
	DefaultExprCurrentTime         DefaultExpr = "current_time"

	// UUID defaults
	DefaultExprUUID           DefaultExpr = "uuid()"
	DefaultExprUUIDSequential DefaultExpr = "uuid_sequential()"

	// Boolean defaults
	DefaultExprTrue  DefaultExpr = "true"
	DefaultExprFalse DefaultExpr = "false"

	// Null
	DefaultExprNull DefaultExpr = "null"
)

func (DefaultExpr) String added in v1.5.7 

func (de DefaultExpr) String() string

type DeltaReader added in v1.2.16 

type DeltaReader struct {
	URI  string
	Duck *DuckDb
	// contains filtered or unexported fields
}

func NewDeltaReader added in v1.2.16 

func NewDeltaReader(uri string, props ...string) (*DeltaReader, error)

func (*DeltaReader) Close added in v1.2.16 

func (r *DeltaReader) Close() error

func (*DeltaReader) Columns added in v1.2.16 

func (r *DeltaReader) Columns() (Columns, error)

func (*DeltaReader) MakeQuery added in v1.2.19 

func (r *DeltaReader) MakeQuery(sc FileStreamConfig) string

type DuckDb added in v1.2.16 

type DuckDb struct {
	Context *g.Context
	Proc    *process.Proc
	// contains filtered or unexported fields
}

DuckDb is a Duck DB compute layer

func NewDuckDb added in v1.2.16 

func NewDuckDb(ctx context.Context, props ...string) *DuckDb

NewDuckDb creates a new DuckDb instance with the given context and properties

func (*DuckDb) AddExtension added in v1.2.16 

func (duck *DuckDb) AddExtension(extension string)

AddExtension adds an extension to the DuckDb instance if it's not already present

func (*DuckDb) AddSecret added in v1.4.11 

func (duck *DuckDb) AddSecret(secret DuckDbSecret)

AddSecret registers a new secret in session (also adds needed extension)

func (*DuckDb) CheckExtension added in v1.4.18 

func (duck *DuckDb) CheckExtension(extension string) (bool, error)

CheckExtension checks if an extension is installed in DuckDB

func (*DuckDb) Close added in v1.2.16 

func (duck *DuckDb) Close() error

Close closes the connection

func (*DuckDb) CopyFormat added in v1.6.4 

func (duck *DuckDb) CopyFormat() (format dbio.FileType, explicit bool, err error)

CopyFormat returns the format that moves data between sling and the DuckDB CLI: arrow (the default) or csv. explicit is false when the default applies.

func (*DuckDb) DataflowToHttpStream added in v1.3.6 

func (duck *DuckDb) DataflowToHttpStream(df *Dataflow, sc StreamConfig) (streamPartChn chan HttpStreamPart, err error)

func (*DuckDb) DefaultCsvConfig added in v1.3.6 

func (duck *DuckDb) DefaultCsvConfig() (config StreamConfig)

func (*DuckDb) Describe added in v1.2.16 

func (duck *DuckDb) Describe(query string) (columns Columns, err error)

Describe returns the columns of a query

func (*DuckDb) EnsureBinDuckDB added in v1.4.23 

func (duck *DuckDb) EnsureBinDuckDB(version string) (binPath string, err error)

EnsureBinDuckDB ensures duckdb binary exists if missing, downloads and uses

func (*DuckDb) Exec added in v1.2.16 

func (duck *DuckDb) Exec(sql string, args ...any) (result sql.Result, err error)

Exec executes a SQL query and returns the result

func (*DuckDb) ExecContext added in v1.2.16 

func (duck *DuckDb) ExecContext(ctx context.Context, sql string, args ...any) (result sql.Result, err error)

ExecContext executes a SQL query with context and returns the result

func (*DuckDb) ExecMultiContext added in v1.2.16 

func (duck *DuckDb) ExecMultiContext(ctx context.Context, sqls ...string) (result sql.Result, err error)

ExecMultiContext executes multiple SQL queries with context and returns the result

func (*DuckDb) GenerateCopyStatement added in v1.2.25 

func (duck *DuckDb) GenerateCopyStatement(fromTable, toLocalPath string, options DuckDbCopyOptions) (sql string, err error)

func (*DuckDb) GenerateCsvColumns added in v1.3.6 

func (duck *DuckDb) GenerateCsvColumns(columns Columns) (colStr string)

func (*DuckDb) GetProp added in v1.2.16 

func (duck *DuckDb) GetProp(key string) string

GetProp retrieves a property value for the DuckDb instance

func (*DuckDb) GetScannerFunc added in v1.2.19 

func (duck *DuckDb) GetScannerFunc(format dbio.FileType) (scanFunc string)

func (*DuckDb) MakeScanQuery added in v1.2.19 

func (duck *DuckDb) MakeScanQuery(format dbio.FileType, uri string, fsc FileStreamConfig) (sql string)

func (*DuckDb) MaxLineSize added in v1.5.25 

func (duck *DuckDb) MaxLineSize(columns Columns) int

MaxLineSize returns the read_csv max_line_size for the given columns. Any column that can hold an unbounded value raises the limit, since one row would otherwise exceed DuckDB's 2 MB default. It is a limit, not an allocation. The `max_line_size` prop overrides it.

func (*DuckDb) Open added in v1.2.16 

func (duck *DuckDb) Open(timeOut ...int) (err error)

Open initializes the DuckDb connection

func (*DuckDb) PrepareFsSecretAndURI added in v1.2.16 

func (duck *DuckDb) PrepareFsSecretAndURI(uri string) string

PrepareFsSecretAndURI prepares the secret configuration from the fs_props and modifies the URI if necessary for different storage types (S3, Google Cloud Storage, Azure Blob Storage). It returns the modified URI string.

The function handles the following storage types: - Local files: Removes the "file://" prefix - S3: Configures AWS credentials and handles Cloudflare R2 storage - Google Cloud Storage: Sets up GCS credentials - Azure Blob Storage: Configures Azure connection string or account name

It uses the DuckDb instance's properties to populate the secret configuration.

func (*DuckDb) Props added in v1.2.16 

func (duck *DuckDb) Props() map[string]string

Props returns all properties of the DuckDb instance as a map

func (*DuckDb) Query added in v1.2.16 

func (duck *DuckDb) Query(sql string, options ...map[string]any) (data Dataset, err error)

Query runs a sql query, returns `Dataset`

func (*DuckDb) QueryContext added in v1.2.16 

func (duck *DuckDb) QueryContext(ctx context.Context, sql string, options ...map[string]any) (data Dataset, err error)

QueryContext runs a sql query with context, returns `Dataset`

func (*DuckDb) Quote added in v1.2.16 

func (duck *DuckDb) Quote(col string) (qName string)

Quote quotes a column name

func (*DuckDb) ReadCsvExpr added in v1.5.26 

func (duck *DuckDb) ReadCsvExpr(uri string, columns Columns) string

ReadCsvExpr returns the read_csv from-expression for the CSV bridge. DuckDB sizes its read buffer as 16 × max_line_size and allocates it eagerly, so the 256MB raise would demand a 4 GiB block and OOM hosts with a smaller memory_limit. An explicit buffer_size caps the allocation at max_line_size, the minimum that fits one line.

func (*DuckDb) SessionFormat added in v1.6.4 

func (duck *DuckDb) SessionFormat() (format dbio.FileType, err error)

SessionFormat returns the format that moves data in and out of the CLI session. The default arrow format changes to csv when the session cannot load the arrow extension (e.g. with no internet access to download it).

func (*DuckDb) SetProp added in v1.2.16 

func (duck *DuckDb) SetProp(key string, value string)

SetProp sets a property for the DuckDb instance

func (*DuckDb) Stream added in v1.2.16 

func (duck *DuckDb) Stream(sql string, options ...map[string]any) (ds *Datastream, err error)

Stream runs a sql query, returns `Datastream`

func (*DuckDb) StreamContext added in v1.2.16 

func (duck *DuckDb) StreamContext(ctx context.Context, sql string, options ...map[string]any) (ds *Datastream, err error)

StreamContext runs a sql query with context, returns `Datastream`

func (*DuckDb) SubmitSQL added in v1.2.19 

func (duck *DuckDb) SubmitSQL(sql string, showChanges bool) (err error)

SubmitSQL submits a sql query to duckdb via stdin

func (*DuckDb) TouchQueryActivity added in v1.6.0 

func (duck *DuckDb) TouchQueryActivity()

TouchQueryActivity resets the stall clock for the in-flight query. Call it only on real evidence of progress.

type DuckDbCopyOptions added in v1.2.25 

type DuckDbCopyOptions struct {
	Format             dbio.FileType
	Compression        CompressorType
	PartitionFields    []PartitionLevel // part_year, part_month, part_day, etc.
	PartitionKey       string
	WritePartitionCols bool
	FileSizeBytes      int64
	GeometryCRS        string  // optional, stamps exported geometry columns with this CRS
	Columns            Columns // optional, used to decode hex-encoded binary and geometry
	HexBinary          bool    // binary columns come as hex text (csv), not as blob (arrow)
}

type DuckDbSecret added in v1.4.11 

type DuckDbSecret struct {
	Type  DuckDbSecretType  `json:"type"`
	Name  string            `json:"name"`
	Props map[string]string `json:"props"`
}

func NewDuckDbSecret added in v1.4.11 

func NewDuckDbSecret(name string, secretType DuckDbSecretType, props map[string]string) DuckDbSecret

func (*DuckDbSecret) AddProp added in v1.4.11 

func (dds *DuckDbSecret) AddProp(key, value string)

func (*DuckDbSecret) Render added in v1.4.11 

func (dds *DuckDbSecret) Render() string

type DuckDbSecretType added in v1.4.11 

type DuckDbSecretType string
const (
	DuckDbSecretTypeUnknown DuckDbSecretType = ""
	DuckDbSecretTypeS3      DuckDbSecretType = "s3"
	DuckDbSecretTypeR2      DuckDbSecretType = "r2"
	DuckDbSecretTypeGCS     DuckDbSecretType = "gcs"
	DuckDbSecretTypeAzure   DuckDbSecretType = "azure"
	DuckDbSecretTypeIceberg DuckDbSecretType = "iceberg"
)

type Encoding added in v1.4.13 

type Encoding string
var (
	EncodingLatin1      Encoding = "latin1"
	EncodingLatin5      Encoding = "latin5"
	EncodingLatin9      Encoding = "latin9"
	EncodingUtf8        Encoding = "utf8"
	EncodingUtf8Bom     Encoding = "utf8_bom"
	EncodingUtf16       Encoding = "utf16"
	EncodingWindows1250 Encoding = "windows1250"
	EncodingWindows1252 Encoding = "windows1252"

	Encodings = []Encoding{
		EncodingLatin1,
		EncodingLatin5,
		EncodingLatin9,
		EncodingUtf8,
		EncodingUtf8Bom,
		EncodingUtf16,
		EncodingWindows1250,
		EncodingWindows1252,
	}
)

func (Encoding) DecodeString added in v1.4.17 

func (e Encoding) DecodeString() string

func (Encoding) EncodeString added in v1.4.17 

func (e Encoding) EncodeString() string

func (Encoding) String added in v1.4.13 

func (e Encoding) String() string

type Evaluator added in v1.4.14 

type Evaluator struct {
	Eval            *goval.Evaluator
	State           map[string]any
	NoComputeKey    string
	VarPrefixes     []string
	KeepMissingExpr bool // allows us to leave any missing sub-expression intact
	AllowNoPrefix   bool
	IgnoreSyntaxErr bool
}

func NewEvaluator added in v1.4.14 

func NewEvaluator(varPrefixes []string, states ...map[string]any) *Evaluator

func (*Evaluator) Check added in v1.4.20 

func (e *Evaluator) Check(expr string) (err error)

func (*Evaluator) ExtractVars added in v1.4.14 

func (e *Evaluator) ExtractVars(expr string) []string

ExtractVars identifies variable references in a string expression, ignoring those inside double quotes. It can recognize patterns like env.VAR, state.VAR, secrets.VAR, and auth.VAR. When AllowNoPrefix is true, it also captures unprefixed variables like MY_VAR, some_value, etc.

func (*Evaluator) FillMissingKeys added in v1.4.25 

func (e *Evaluator) FillMissingKeys(stateMap map[string]any, varsToCheck []string) map[string]any

func (*Evaluator) FindMatches added in v1.5.1 

func (e *Evaluator) FindMatches(inputStr string) (expressions []string, err error)

FindMatches parses the input string and extracts expressions within curly braces, properly handling nested brackets and quoted strings. Returns an error if brackets are unbalanced.

func (*Evaluator) RenderAny added in v1.4.14 

func (e *Evaluator) RenderAny(input any, extras ...map[string]any) (output any, err error)

func (*Evaluator) RenderPayload added in v1.4.14 

func (e *Evaluator) RenderPayload(val any, extras ...map[string]any) (newVal any, err error)

func (*Evaluator) RenderString added in v1.4.14 

func (e *Evaluator) RenderString(val any, extras ...map[string]any) (newVal string, err error)

type Excel added in v1.2.2 

type Excel struct {
	File   *excelize.File
	Sheets []string
	Path   string
	// contains filtered or unexported fields
}

Excel represent an Excel object pointing to its file

func NewExcel added in v1.2.2 

func NewExcel() (xls *Excel)

NewExcel creates a new excel file

func NewExcelFromFile added in v1.2.2 

func NewExcelFromFile(path string) (xls *Excel, err error)

NewExcelFromFile return a new Excel instance from a local file

func NewExcelFromReader added in v1.2.2 

func NewExcelFromReader(reader io.Reader, opts ...excelize.Options) (xls *Excel, err error)

NewExcelFromReader return a new Excel instance from a reader

func (*Excel) GetDataset added in v1.2.2 

func (xls *Excel) GetDataset(sheet string) (data Dataset)

GetDataset returns a dataset of the provided sheet

func (*Excel) GetDatasetFromRange added in v1.2.2 

func (xls *Excel) GetDatasetFromRange(sheet, cellRange string) (data Dataset, err error)

GetDatasetFromRange returns a dataset of the provided sheet / range cellRange examples:

  • `$AH$13:$AI$20` or `AH13:AI20` - standard cell range
  • `A:E` - column range (all rows)
  • `9:15` - row range (detects columns with data)
  • `A4:C` - partial range (extends to last row)

func (*Excel) NumberToTitle added in v1.5.6 

func (xls *Excel) NumberToTitle(n int) string

NumberToTitle converts a 0-based column index to Excel column letters. For example: 0 -> "A", 25 -> "Z", 26 -> "AA"

func (*Excel) RefreshSheets added in v1.2.2 

func (xls *Excel) RefreshSheets() (err error)

RefreshSheets refresh sheet index data

func (*Excel) TitleToNumber added in v1.4.24 

func (xls *Excel) TitleToNumber(s string) int

TitleToNumber provides a function to convert Excel sheet column title to int (this function doesn't do value check currently). For example convert AK and ak to column title 36: 

excelize.TitleToNumber("AK")
excelize.TitleToNumber("ak")

func (*Excel) WriteSheet added in v1.2.2 

func (xls *Excel) WriteSheet(shtName string, ds *Datastream, mode string) (err error)

WriteSheet write a datastream into a sheet mode can be: `new`, `append` or `overwrite`. Default is `new`

func (*Excel) WriteToFile added in v1.2.2 

func (xls *Excel) WriteToFile(path string) (err error)

WriteToFile write to a file

func (*Excel) WriteToWriter added in v1.2.2 

func (xls *Excel) WriteToWriter(w io.Writer) (err error)

WriteToWriter write to a provided writer

type FileStreamConfig added in v1.2.19 

type FileStreamConfig struct {
	Limit            int               `json:"limit"`
	Select           []string          `json:"select"`
	SQL              string            `json:"sql"`
	Format           dbio.FileType     `json:"format"`
	IncrementalKey   string            `json:"incremental_key"`
	IncrementalValue string            `json:"incremental_value"`
	FileSelect       []string          `json:"file_select"`     // a list of files to include.
	DuckDBFilename   bool              `json:"duckdb_filename"` // stream URL
	SchemaOnly       bool              `json:"schema_only"`
	Props            map[string]string `json:"props"`
}

func (*FileStreamConfig) AddsFilenameColumn added in v1.6.0 

func (sc *FileStreamConfig) AddsFilenameColumn() bool

AddsFilenameColumn tells whether the scan query appends a filename column, which becomes _sling_stream_url. A custom sql must ask for it with the {filename_expr} placeholder, otherwise its own last column is the last one.

func (*FileStreamConfig) GetProp added in v1.3.5 

func (sc *FileStreamConfig) GetProp(key string) string

func (*FileStreamConfig) SetProp added in v1.3.5 

func (sc *FileStreamConfig) SetProp(key, val string)

func (*FileStreamConfig) ShouldUseDuckDB added in v1.2.19 

func (sc *FileStreamConfig) ShouldUseDuckDB() bool

type ForeignKeyInfo added in v1.5.7 

type ForeignKeyInfo struct {
	ConstraintName   string `json:"constraint_name"`
	ColumnName       string `json:"column_name,omitempty"`
	ReferencedSchema string `json:"referenced_schema"`
	ReferencedTable  string `json:"referenced_table"`
	ReferencedColumn string `json:"referenced_column"`
	OnDelete         string `json:"on_delete,omitempty"`
	OnUpdate         string `json:"on_update,omitempty"`
}

ForeignKeyInfo represents a foreign key relationship

type GoogleSheet added in v1.2.2 

type GoogleSheet struct {
	Sheets        []string
	SpreadsheetID string
	// contains filtered or unexported fields
}

GoogleSheet represent a Google Sheet object

func NewGoogleSheet added in v1.2.2 

func NewGoogleSheet(props ...string) (ggs *GoogleSheet, err error)

NewGoogleSheet is a blank spreadsheet title is the new spreadsheet title

func NewGoogleSheetFromURL added in v1.2.2 

func NewGoogleSheetFromURL(urlStr string, props ...string) (ggs *GoogleSheet, err error)

NewGoogleSheetFromURL return a new GoogleSheet instance from a provided url

func (*GoogleSheet) DeleteSheet added in v1.2.2 

func (ggs *GoogleSheet) DeleteSheet(shtName string) (err error)

func (*GoogleSheet) GetDataset added in v1.2.2 

func (ggs *GoogleSheet) GetDataset(shtName string) (data Dataset, err error)

GetDataset returns a dataset of the sheet

func (*GoogleSheet) GetDatasetFromRange added in v1.2.2 

func (ggs *GoogleSheet) GetDatasetFromRange(shtName, cellRange string) (data Dataset, err error)

GetDatasetFromRange returns a dataset from the specified range

func (*GoogleSheet) RefreshSheets added in v1.2.2 

func (ggs *GoogleSheet) RefreshSheets() (err error)

RefreshSheets refreshes sheets data

func (*GoogleSheet) URL added in v1.2.2 

func (ggs *GoogleSheet) URL() string

func (*GoogleSheet) WriteSheet added in v1.2.2 

func (ggs *GoogleSheet) WriteSheet(shtName string, ds *Datastream, mode string) (err error)

WriteSheet write a datastream into a sheet mode can be: `new`, `append` or `overwrite`. Default is `new`

type GzipCompressor 

type GzipCompressor struct {
	Compressor
	// contains filtered or unexported fields
}

func (*GzipCompressor) Compress 

func (cp *GzipCompressor) Compress(reader io.Reader) io.Reader

Compress uses gzip to compress

func (*GzipCompressor) Decompress 

func (cp *GzipCompressor) Decompress(reader io.Reader) (gReader io.Reader, err error)

Decompress uses gzip to decompress if it is gzip. Otherwise return same reader

func (*GzipCompressor) Suffix 

func (cp *GzipCompressor) Suffix() string

type HttpStreamPart added in v1.3.6 

type HttpStreamPart struct {
	Index    int
	FromExpr string
	Columns  Columns

	// Cancel unblocks the producer goroutine in DataflowToHttpStream. Consumers
	// MUST defer it when they range over streamPartChn: leaving the loop early
	// (on an insert error) otherwise strands the producer on an unbuffered send
	// with no reader, hanging the process.
	Cancel context.CancelFunc
}

type IcebergReader added in v1.2.16 

type IcebergReader struct {
	URI  string
	Duck *DuckDb
	// contains filtered or unexported fields
}

func NewIcebergReader added in v1.2.16 

func NewIcebergReader(uri string, props ...string) (*IcebergReader, error)

func (*IcebergReader) Close added in v1.2.16 

func (i *IcebergReader) Close() error

func (*IcebergReader) Columns added in v1.2.16 

func (r *IcebergReader) Columns() (Columns, error)

func (*IcebergReader) MakeQuery added in v1.2.19 

func (r *IcebergReader) MakeQuery(sc FileStreamConfig) string

type IndexColumn added in v1.5.20 

type IndexColumn struct {
	Name string `json:"name"`           // column name or expression
	Sort string `json:"sort,omitempty"` // "asc" / "desc" / ""
}

IndexColumn is one column (or per-column expression) within an index.

type IndexDef added in v1.5.20 

type IndexDef struct {
	Name       string        `json:"name,omitempty"`       // auto-generated if empty
	Columns    []IndexColumn `json:"columns,omitempty"`    // mutually exclusive with Expression
	Expression string        `json:"expression,omitempty"` // raw SQL for expression indexes
	Unique     bool          `json:"unique,omitempty"`
	Where      string        `json:"where,omitempty"`
	Type       string        `json:"type,omitempty"` // btree / hash / gin / ...
	Include    []string      `json:"include,omitempty"`

	// Priority is used only while merging column-inline composite indexes (same
	// index Name across multiple columns); it orders the columns and is not part
	// of the rendered DDL.
	Priority int `json:"priority,omitempty"`
}

IndexDef is the unified internal representation of an index, populated both from the columns: inline modifier DSL and from target_options.table_keys.index. DDL generation reads from this single source of truth.

func (IndexDef) ColumnNames added in v1.5.20 

func (id IndexDef) ColumnNames() (names []string)

ColumnNames returns the column/expression names of the index.

type IndexInfo added in v1.5.7 

type IndexInfo struct {
	Name     string   `json:"name"`
	Columns  []string `json:"columns"`
	IsUnique bool     `json:"is_unique"`
}

IndexInfo represents an index

type Iterator 

type Iterator struct {
	Row          []any
	Reprocess    chan []any
	IsCasted     bool
	RowIsCasted  bool
	Counter      uint64
	StreamRowNum uint64
	Context      *g.Context
	Closed       bool
	// contains filtered or unexported fields
}

Iterator is the row provider for a datastream

func (*Iterator) BelowEqualIncrementalVal added in v1.2.10 

func (it *Iterator) BelowEqualIncrementalVal() bool

BelowEqualIncrementalVal evaluates the incremental value against the incrementalCol this is used when the stream is a file with incremental mode (unable to filter at source like a database) it.incrementalVal and it.incrementalColI need to be set

func (*Iterator) Ds 

func (it *Iterator) Ds() *Datastream

type JsonColumnTyping added in v1.4.7 

type JsonColumnTyping struct {
	AsText bool `json:"as_text,omitempty" yaml:"as_text,omitempty"`
}

JsonColumnTyping contains json type mapping configurations

func (*JsonColumnTyping) Apply added in v1.4.7 

func (jct *JsonColumnTyping) Apply(col *Column)

type KeyType 

type KeyType string
const (
	AggregateKey    KeyType = "aggregate"
	ClusterKey      KeyType = "cluster"
	DistributionKey KeyType = "distribution"
	DuplicateKey    KeyType = "duplicate"
	HashKey         KeyType = "hash"
	IndexKey        KeyType = "index"
	PartitionKey    KeyType = "partition"
	PrimaryKey      KeyType = "primary"
	SortKey         KeyType = "sort"
	UniqueKey       KeyType = "unique"
	UpdateKey       KeyType = "update"
)

func (KeyType) MetadataKey added in v1.2.15 

func (kt KeyType) MetadataKey() string

type KeyValue 

type KeyValue struct {
	Key   string `json:"key"`
	Value any    `json:"value"`
}

type MetaColumn added in v1.6.4 

type MetaColumn struct {
	Column Column
	Value  func(rowNum int64) any
}

MetaColumn is one metadata column a stream appends to every record. Value is called once per row, with the stream's running row number.

type Metadata 

type Metadata struct {
	StreamURL KeyValue `json:"stream_url"`
	SyncedAt  KeyValue `json:"synced_at"`
	SyncedOp  KeyValue `json:"synced_op"`
	SyncedSeq KeyValue `json:"synced_seq"`
	RowNum    KeyValue `json:"row_num"`
	RowID     KeyValue `json:"row_id"`
	ExecID    KeyValue `json:"exec_id"`
}

func (*Metadata) AsMap 

func (m *Metadata) AsMap() map[string]any

AsMap return as map

type NoneCompressor 

type NoneCompressor struct {
	Compressor
	// contains filtered or unexported fields
}

func (*NoneCompressor) Compress 

func (cp *NoneCompressor) Compress(reader io.Reader) io.Reader

func (*NoneCompressor) Decompress 

func (cp *NoneCompressor) Decompress(reader io.Reader) (gReader io.Reader, err error)

func (*NoneCompressor) Suffix 

func (cp *NoneCompressor) Suffix() string

type Parquet 

type Parquet struct {
	Path   string
	Reader *parquet.Reader
	Data   *Dataset
	// contains filtered or unexported fields
}

Parquet is a parquet object

func NewParquetReader added in v1.1.7 

func NewParquetReader(reader io.ReaderAt, columns Columns) (p *Parquet, err error)

func (*Parquet) Columns 

func (p *Parquet) Columns() Columns

type ParquetArrowReader added in v1.1.6 

type ParquetArrowReader struct {
	Path    string
	Reader  *pqarrow.FileReader
	File    *os.File
	Data    *Dataset
	Context *g.Context
	Memory  memory.Allocator
	// contains filtered or unexported fields
}

ParquetArrowReader is a parquet reader object using arrow v18

func NewParquetArrowReader added in v1.1.7 

func NewParquetArrowReader(reader *os.File, selected []string) (p *ParquetArrowReader, err error)

func (*ParquetArrowReader) Columns added in v1.1.6 

func (p *ParquetArrowReader) Columns() Columns

type ParquetArrowWriter added in v1.1.6 

type ParquetArrowWriter struct {
	Writer *pqarrow.FileWriter
	// contains filtered or unexported fields
}

func NewParquetArrowWriter added in v1.1.6 

func NewParquetArrowWriter(w io.Writer, columns Columns, codec compress.Compression) (p *ParquetArrowWriter, err error)

func NewParquetArrowWriterFromSchema added in v1.6.4 

func NewParquetArrowWriterFromSchema(w io.Writer, schema *arrow.Schema, codec compress.Compression) (p *ParquetArrowWriter, err error)

NewParquetArrowWriterFromSchema returns a Parquet writer that writes whole records, for the arrow lane. There are no per-column builders: the record buffers go straight to pqarrow.

func (*ParquetArrowWriter) Close added in v1.1.6 

func (p *ParquetArrowWriter) Close() error

func (*ParquetArrowWriter) Columns added in v1.1.6 

func (p *ParquetArrowWriter) Columns() Columns

func (*ParquetArrowWriter) RowGroupBytes added in v1.6.4 

func (p *ParquetArrowWriter) RowGroupBytes() int64

RowGroupBytes returns the bytes buffered in the current row group.

func (*ParquetArrowWriter) WriteRecord added in v1.6.4 

func (p *ParquetArrowWriter) WriteRecord(rec arrow.RecordBatch) error

WriteRecord writes one record to the Parquet file. The record must carry the writer's schema. Small records are grouped into one row group by WriteBuffered; a new row group starts once the current one passes parquetArrowRowGroupBytes.

func (*ParquetArrowWriter) WriteRow added in v1.1.6 

func (p *ParquetArrowWriter) WriteRow(row []any) error

type ParquetDuckDb added in v1.2.16 

type ParquetDuckDb struct {
	URI  string
	Duck *DuckDb
	// contains filtered or unexported fields
}

func NewParquetReaderDuckDb added in v1.2.16 

func NewParquetReaderDuckDb(uri string, props ...string) (*ParquetDuckDb, error)

func (*ParquetDuckDb) Close added in v1.2.16 

func (r *ParquetDuckDb) Close() error

func (*ParquetDuckDb) Columns added in v1.2.16 

func (r *ParquetDuckDb) Columns() (Columns, error)

func (*ParquetDuckDb) MakeQuery added in v1.2.19 

func (r *ParquetDuckDb) MakeQuery(sc FileStreamConfig) string

type ParquetWriter added in v1.1.7 

type ParquetWriter struct {
	Writer    *parquet.Writer
	WriterMap *parquet.GenericWriter[map[string]any]
	// contains filtered or unexported fields
}

func NewParquetWriter added in v1.1.7 

func NewParquetWriter(w io.Writer, columns Columns, codec compress.Codec) (p *ParquetWriter, err error)

func NewParquetWriterMap added in v1.2.12 

func NewParquetWriterMap(w io.Writer, columns Columns, codec compress.Codec) (p *ParquetWriter, err error)

func (*ParquetWriter) Close added in v1.1.7 

func (pw *ParquetWriter) Close() error

func (*ParquetWriter) WriteRec added in v1.2.12 

func (pw *ParquetWriter) WriteRec(row []any) error

func (*ParquetWriter) WriteRow added in v1.1.7 

func (pw *ParquetWriter) WriteRow(row []any) error

type PartitionLevel added in v1.3.4 

type PartitionLevel string
const (
	PartitionLevelSecond    PartitionLevel = "second"
	PartitionLevelMinute    PartitionLevel = "minute"
	PartitionLevelHour      PartitionLevel = "hour"
	PartitionLevelDay       PartitionLevel = "day"
	PartitionLevelWeek      PartitionLevel = "week"
	PartitionLevelYearMonth PartitionLevel = "year_month"
	PartitionLevelMonth     PartitionLevel = "month"
	PartitionLevelYear      PartitionLevel = "year"
)

func ExtractPartitionFields added in v1.3.4 

func ExtractPartitionFields(path string) (levels []PartitionLevel)

ExtractPartitionFields extract the partition fields from the given path

func GetLowestPartTimeLevel added in v1.3.6 

func GetLowestPartTimeLevel(url string) (PartitionLevel, error)

func (PartitionLevel) IsValid added in v1.3.4 

func (level PartitionLevel) IsValid() bool

func (PartitionLevel) Str added in v1.3.6 

func (level PartitionLevel) Str() string

func (PartitionLevel) TruncateTime added in v1.3.4 

func (level PartitionLevel) TruncateTime(t time.Time) (time.Time, error)

type Queue added in v1.4.11 

type Queue struct {
	Name   string        `json:"name"`
	Path   string        `json:"path"`
	File   *os.File      `json:"-"`
	Reader *bufio.Reader `json:"-"`
	Writer *bufio.Writer `json:"-"`
	// contains filtered or unexported fields
}

func NewQueue added in v1.4.11 

func NewQueue(name string) (q *Queue, err error)

NewQueue creates a new queue with a temporary file

func (*Queue) Append added in v1.4.11 

func (q *Queue) Append(data any) error

Append writes a line to the queue

func (*Queue) Close added in v1.4.11 

func (q *Queue) Close() error

Close closes and optionally removes the queue file

func (*Queue) Flush added in v1.5.20 

func (q *Queue) Flush() error

Flush flushes and syncs buffered writes to disk; call before MarkDone.

func (*Queue) IsDone added in v1.5.20 

func (q *Queue) IsDone() bool

IsDone reports whether the producer has signaled completion.

func (*Queue) MarkDone added in v1.5.20 

func (q *Queue) MarkDone() error

MarkDone writes the done-sentinel; call after the producer flushed all writes.

func (*Queue) NewReader added in v1.5.20 

func (q *Queue) NewReader() (*QueueReader, error)

NewReader opens a fresh tailing reader at the start of the file; caller Closes it.

func (*Queue) Next added in v1.4.11 

func (q *Queue) Next() (any, bool, error)

Next reads the next line from the queue Returns the line, a boolean indicating if there are more lines, and any error

func (*Queue) Reset added in v1.4.11 

func (q *Queue) Reset() error

Reset positions the reader at the beginning of the file

type QueueReader added in v1.5.20 

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

QueueReader is an independent tailing reader over a Queue file. Each consumer gets its own fd and offset, so all consumers see every record (broadcast). Polls for growth + watches the done-sentinel (no OS file-watch); cross-platform.

func (*QueueReader) Close added in v1.5.20 

func (qr *QueueReader) Close() error

Close releases the reader's file descriptor.

func (*QueueReader) Next added in v1.5.20 

func (qr *QueueReader) Next(ctx context.Context) (data any, hasMore bool, err error)

Next returns the next record, blocking (polling) at EOF until the producer appends more or signals done. Returns hasMore=false when done and drained, or when ctx is cancelled (so a failing sibling unblocks all readers promptly).

type ReaderReady added in v1.2.6 

type ReaderReady struct {
	Reader io.Reader
	URI    string

	// Open obtains the body. It can run again to replace a body that a
	// remote store reset while the body sat idle (e.g. in a prefetch queue).
	Open func() (io.Reader, error)
	// contains filtered or unexported fields
}

func (*ReaderReady) Close added in v1.5.25 

func (rr *ReaderReady) Close() error

Close closes an unused or finished body. Safe to call more than once.

func (*ReaderReady) GetReader added in v1.5.25 

func (rr *ReaderReady) GetReader() (io.Reader, error)

GetReader returns the reader. It opens on first use. A failed Open is retried on the next call. When Open is set, the returned reader recovers from a reset body: it reopens, skips the bytes already delivered, and resumes (see Read).

func (*ReaderReady) Prefetch added in v1.5.25 

func (rr *ReaderReady) Prefetch()

Prefetch opens the body ahead of consumption. An error is not returned: GetReader retries the open at the point of use.

func (*ReaderReady) Read added in v1.5.25 

func (rr *ReaderReady) Read(p []byte) (n int, err error)

Read delegates to the body. On a read error before any byte of this call, it reopens the body, skips the bytes already delivered, and retries. This recovers a body that a remote store reset while it sat idle.

type RecNode 

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

func NewRecNode 

func NewRecNode(cols Columns, optional bool) *RecNode

func (*RecNode) Compression 

func (rn *RecNode) Compression() compress.Codec

func (*RecNode) Encoding 

func (rn *RecNode) Encoding() encoding.Encoding

func (*RecNode) Fields 

func (rn *RecNode) Fields() []parquet.Field

func (*RecNode) GoType 

func (rn *RecNode) GoType() reflect.Type

func (*RecNode) ID 

func (rn *RecNode) ID() int

func (*RecNode) Leaf 

func (rn *RecNode) Leaf() bool

func (*RecNode) Optional 

func (rn *RecNode) Optional() bool

func (*RecNode) Repeated 

func (rn *RecNode) Repeated() bool

func (*RecNode) Required 

func (rn *RecNode) Required() bool

func (*RecNode) String 

func (rn *RecNode) String() string

func (*RecNode) Type 

func (rn *RecNode) Type() parquet.Type

type Record 

type Record struct {
	Columns *Columns
	Values  []any
}

type RecordStream added in v1.6.4 

type RecordStream struct {
	Schema  *arrow.Schema
	Columns Columns
	// contains filtered or unexported fields
}

RecordStream carries Arrow record batches from a source to a sink. It is the lane's replacement for BatchChan: the sink pulls records instead of rows, and no []any row is built on the way.

Ownership moves at Push: the caller retains, the stream releases. Next hands the record to the consumer, which releases it. Peek does not consume.

func NewRecordStream added in v1.6.4 

func NewRecordStream(ctx *g.Context, lane ArrowLane, schema *arrow.Schema, size int) *RecordStream

NewRecordStream returns a stream that carries records in the given schema. The lane must not be nil: the gate never lets a nil lane reach here, so a nil lane marks a wiring bug.

func (*RecordStream) Close added in v1.6.4 

func (rs *RecordStream) Close(err error)

Close ends the stream. Only the producer calls it: the consumer side calls Fail. The first error wins.

func (*RecordStream) Drain added in v1.6.4 

func (rs *RecordStream) Drain()

Drain releases every record still queued or peeked, and keeps draining until the producer closes the stream. It is idempotent.

func (*RecordStream) Err added in v1.6.4 

func (rs *RecordStream) Err() error

Err returns the first error of the stream.

func (*RecordStream) Fail added in v1.6.4 

func (rs *RecordStream) Fail(err error)

Fail records an error from any goroutine and unblocks a producer that is waiting on the record channel. The producer then closes the stream.

func (*RecordStream) Lane added in v1.6.4 

func (rs *RecordStream) Lane() ArrowLane

Lane returns the engine of the stream.

func (*RecordStream) Max added in v1.6.4 

func (rs *RecordStream) Max() (val int64, ok bool)

Max returns the tracked maximum, as unix micro for date and timestamp columns, and ok = false when nothing was tracked or found.

func (*RecordStream) MaxString added in v1.6.4 

func (rs *RecordStream) MaxString() (val string, ok bool)

MaxString returns the tracked maximum of a string update-key column, and ok = false when no string was tracked.

func (*RecordStream) Next added in v1.6.4 

func (rs *RecordStream) Next() (rec arrow.RecordBatch, ok bool)

Next returns the next record, which the consumer owns and releases.

func (*RecordStream) Normalize added in v1.6.4 

func (rs *RecordStream) Normalize(target *arrow.Schema) (*RecordStream, error)

Normalize returns a stream whose records carry the target schema. It is lazy: the lane runs once per record. Returns the receiver when the schema already matches.

func (*RecordStream) NullCounts added in v1.6.4 

func (rs *RecordStream) NullCounts() []int64

NullCounts returns the per-column null count of the records taken so far.

func (*RecordStream) Peek added in v1.6.4 

func (rs *RecordStream) Peek() (arrow.RecordBatch, bool)

Peek returns the first record without consuming it. The stream keeps ownership, so the caller must not release it.

Peek waits for the producer's first record, so it must not hold rs.mu: the producer takes rs.mu on every Push, and holding it here would deadlock the pair. peekMu serializes peekers instead, so one record is held at most.

func (*RecordStream) Project added in v1.6.4 

func (rs *RecordStream) Project(cols Columns) (*RecordStream, error)

Project returns a stream whose records carry the names and order of cols. It is lazy: the lane runs once per record. Returns the receiver when the names and order already match.

func (*RecordStream) Push added in v1.6.4 

func (rs *RecordStream) Push(rec arrow.RecordBatch) error

Push sends a record to the consumer. The caller retains the record; the stream releases it. Push blocks until the consumer takes it, which is the lane's backpressure, and returns when the context is done.

func (*RecordStream) Reader added in v1.6.4 

func (rs *RecordStream) Reader() array.RecordReader

Reader returns the stream as an array.RecordReader, for the sinks that ingest a reader (adbc.IngestStream).

func (*RecordStream) Relabel added in v1.6.4 

func (rs *RecordStream) Relabel(schema *arrow.Schema) (*RecordStream, error)

Relabel returns a stream whose records carry schema. Only the field labels (metadata) may differ: the names and types must match, so no value changes. Returns the receiver when the schema already matches.

func (*RecordStream) SampleRows added in v1.6.4 

func (rs *RecordStream) SampleRows(n int) [][]any

SampleRows returns up to n rows of the first record as []any, for the datastream buffer (DDL and the sample checks). It does not consume.

func (*RecordStream) SetMetaColumns added in v1.6.4 

func (rs *RecordStream) SetMetaColumns(cols []MetaColumn) error

SetMetaColumns makes the stream append the metadata columns to every record, before the transforms run, as the row path does. Call it before the stream is consumed: the datastream sample reads through Peek, so the first record carries them too, and the columns the stream reports include them.

func (*RecordStream) SetOnTake added in v1.6.4 

func (rs *RecordStream) SetOnTake(f func(rec arrow.RecordBatch))

SetOnTake sets the hook the stream calls when a record is taken by the consumer. The datastream counts rows and bytes there.

func (*RecordStream) SetTransform added in v1.6.4 

func (rs *RecordStream) SetTransform(stages []map[string]string, sp *StreamProcessor) error

SetTransform makes the stream evaluate the stage list on every record, with the lane's engine. Call it before the stream is consumed: the datastream sample reads through Peek, so the first record is transformed too, and the columns and schema the stream reports follow the transform.

The stream processor supplies the transform functions (the same map the row path uses), so both paths evaluate the same expressions.

func (*RecordStream) TrackMax added in v1.6.4 

func (rs *RecordStream) TrackMax(colIdx int)

TrackMax tracks the maximum value of one column, for incremental state. Call it before the stream is consumed.

func (*RecordStream) TrackedMax added in v1.6.4 

func (rs *RecordStream) TrackedMax() (colIdx int, val int64, ok bool)

TrackedMax returns the tracked column index and its maximum value.

func (*RecordStream) TrackedMaxString added in v1.6.4 

func (rs *RecordStream) TrackedMaxString() (colIdx int, val string, ok bool)

TrackedMaxString returns the tracked column index and its string maximum.

type RecordTransform added in v1.6.4 

type RecordTransform interface {
	// Transform evaluates the stages on one record and returns the new record
	// with the columns that follow it. The column types it reports stay fixed
	// for the stream: a later record whose result moves to another type class
	// is an error, where the row path would coerce it.
	Transform(rec arrow.RecordBatch, cols Columns) (arrow.RecordBatch, Columns, error)
}

RecordTransform evaluates a stage list on records. The caller owns the result; the input record is not consumed.

type SAS 

type SAS struct {
	Path   string
	Reader *datareader.SAS7BDAT
	Data   *Dataset
	// contains filtered or unexported fields
}

SAS is a sas7bdat object

func NewSASStream 

func NewSASStream(reader io.ReadSeeker, columns Columns) (s *SAS, err error)

func (*SAS) Columns 

func (s *SAS) Columns() Columns

type SSHClient 

type SSHClient struct {
	Host        string
	Port        int
	User        string
	Password    string
	TgtHost     string
	TgtPort     int
	PrivateKey  string
	Passphrase  string
	Options     SSHOptions
	HostKeyAddr string // verify the host key for this address instead of Host:Port (e.g. through a tunnel)
	Err         error
	// contains filtered or unexported fields
}

SSHClient is a client to connect to a ssh server with the main goal of forwarding ports

func (*SSHClient) Close 

func (s *SSHClient) Close()

Close stops the client connection

func (*SSHClient) Connect 

func (s *SSHClient) Connect() (err error)

Connect connects to the server

func (*SSHClient) NewSession added in v1.5.1 

func (s *SSHClient) NewSession() (*ssh.Session, error)

NewSession creates a new SSH session

func (*SSHClient) OpenPortForward 

func (s *SSHClient) OpenPortForward() (localPort int, err error)

OpenPortForward forwards the port as specified

func (*SSHClient) SftpClient 

func (s *SSHClient) SftpClient() (sftpClient *sftp.Client, err error)

SftpClient returns an SftpClient

type SSHHostKey added in v1.6.5 

type SSHHostKey struct {
	Keys       []string // accepted fingerprints (SHA256:...) or public keys
	KnownHosts string   // known_hosts file to read, in addition to ~/.ssh/known_hosts
	Strict     *bool    // nil: warn when the key is not verified
}

SSHHostKey verifies the host key of an SSH server

func NewSSHHostKey added in v1.6.5 

func NewSSHHostKey(keys, knownHosts, strict string) SSHHostKey

NewSSHHostKey creates an SSHHostKey from the values of the ssh_host_key, ssh_known_hosts and ssh_strict_host_key properties

func (SSHHostKey) Callback added in v1.6.5 

func (hk SSHHostKey) Callback(addr string) (callback ssh.HostKeyCallback, algorithms []string, err error)

Callback returns the host key callback for the server at addr. It also returns the host key algorithms to request, so that the server sends a known key type. A key in ssh_host_key or in a known_hosts file is trusted.

type SSHOptions added in v1.6.5 

type SSHOptions struct {
	HostKey      SSHHostKey
	Ciphers      []string // nil: the default list
	KeyExchanges []string // nil: the default list
}

SSHOptions are the ssh_* settings of a connection. They apply to the SSH tunnel server and to the SFTP server.

func NewSSHOptions added in v1.6.5 

func NewSSHOptions(getProp func(key string) string) SSHOptions

NewSSHOptions creates SSHOptions from the ssh_* connection properties

func (SSHOptions) Config added in v1.6.5 

func (o SSHOptions) Config() (config ssh.Config, err error)

Config returns the algorithms to offer. The default list puts legacy algorithms after the secure ones, so that old servers can connect.

type Selector added in v1.5.1 

type Selector struct {
	ConnType dbio.Type
	// contains filtered or unexported fields
}

Selector provides cached field selection, exclusion, and renaming. Build via NewSelector; call Apply per field, or OrderFields to lay out the final column list following expression order.

func NewSelector added in v1.5.1 

func NewSelector(selectExprs []string, casing ColumnCasing) *Selector

NewSelector creates a Selector from select expressions. All field matching is case-insensitive.

func (*Selector) Apply added in v1.5.1 

func (s *Selector) Apply(name string) (new string, included bool)

Apply returns the (possibly renamed) field name and whether it's included. Priority: rename > exclude > glob exclude > All > include > glob include. Results are cached per field.

func (*Selector) OrderFields added in v1.5.19 

func (s *Selector) OrderFields(fields []string) []string

OrderFields lays out final output names following expression order. Source-order is preserved within each `*` or glob expansion. Exact names pin to their list position; renames keep source-order position; `*`/globs skip pins and excluded names. `excludeOnly` selectors behave as if `*` were prepended.

type Shaper 

type Shaper struct {
	Func       func([]any) []any
	SrcColumns Columns
	TgtColumns Columns
	ColMap     map[int]int
}

type SnappyCompressor 

type SnappyCompressor struct {
	Compressor
	// contains filtered or unexported fields
}

func (*SnappyCompressor) Compress 

func (cp *SnappyCompressor) Compress(reader io.Reader) io.Reader

Compress uses gzip to compress

func (*SnappyCompressor) Decompress 

func (cp *SnappyCompressor) Decompress(reader io.Reader) (sReader io.Reader, err error)

Decompress uses gzip to decompress if it is gzip. Otherwise return same reader

func (*SnappyCompressor) Suffix 

func (cp *SnappyCompressor) Suffix() string

type StreamConfig added in v1.1.15 

type StreamConfig struct {
	EmptyAsNull      bool           `json:"empty_as_null"`
	Header           bool           `json:"header"`
	Compression      CompressorType `json:"compression"` // AUTO | ZIP | GZIP | SNAPPY | NONE
	NullIf           string         `json:"null_if"`
	NullAs           string         `json:"null_as"`
	DatetimeFormat   string         `json:"datetime_format"`
	SkipBlankLines   bool           `json:"skip_blank_lines"`
	SkipLines        int            `json:"skip_lines"`
	Format           dbio.FileType  `json:"format"`
	Delimiter        string         `json:"delimiter"`
	Escape           string         `json:"escape"`
	Quote            string         `json:"quote"`
	FileMaxRows      int64          `json:"file_max_rows"`
	FileMaxBytes     int64          `json:"file_max_bytes"`
	BatchLimit       int64          `json:"batch_limit"`
	BatchMaxDuration time.Duration  `json:"batch_max_duration"`
	MaxDecimals      int            `json:"max_decimals"`
	Flatten          int            `json:"flatten"`
	FieldsPerRec     int            `json:"fields_per_rec"`
	Jmespath         string         `json:"jmespath"`
	Jq               string         `json:"jq"`
	Select           []string       `json:"select"` // applied to JSON-family streams
	Sheet            string         `json:"sheet"`
	ColumnCasing     ColumnCasing   `json:"column_casing"`
	ColumnTyping     ColumnTyping   `json:"column_typing"`
	TargetType       dbio.Type      `json:"target_type"`
	DeleteFile       bool           `json:"delete"` // whether to delete before writing
	BoolAsInt        bool           `json:"-"`
	EscapeBackslash  bool           `json:"-"`       // escape backslashes for MySQL LOAD DATA
	BinaryAsHex      bool           `json:"-"`       // hex-encode binary in CSV
	Columns          Columns        `json:"columns"` // list of column types. Can be partial list! likely is!
	Transforms       Transform

	Map map[string]string `json:"-"`
}

func DefaultStreamConfig added in v1.1.15 

func DefaultStreamConfig() StreamConfig

func LoaderStreamConfig added in v1.2.22 

func LoaderStreamConfig(header bool) StreamConfig

func (*StreamConfig) ToMap added in v1.2.22 

func (sc *StreamConfig) ToMap() map[string]string

type StreamProcessor 

type StreamProcessor struct {
	N uint64

	Config StreamConfig
	// contains filtered or unexported fields
}

StreamProcessor processes rows and values

func NewStreamProcessor 

func NewStreamProcessor() *StreamProcessor

NewStreamProcessor returns a new StreamProcessor

func (*StreamProcessor) CastRow 

func (sp *StreamProcessor) CastRow(row []any, columns Columns) []any

CastRow casts each value of a row slows down processing about 40%?

func (*StreamProcessor) CastToBool added in v1.2.12 

func (sp *StreamProcessor) CastToBool(i any) (b bool, err error)

CastToBool converts interface to bool

func (*StreamProcessor) CastToString 

func (sp *StreamProcessor) CastToString(val any) (valString string)

func (*StreamProcessor) CastToStringCSV added in v1.4.17 

func (sp *StreamProcessor) CastToStringCSV(i int, val any, valType ...ColumnType) string

CastToStringCSV to string. used for csv writing slows processing down 5% with upstream CastRow or 35% without upstream CastRow

func (*StreamProcessor) CastToStringE added in v1.4.17 

func (sp *StreamProcessor) CastToStringE(val any) (valString string, err error)

func (*StreamProcessor) CastToStringSafeMask added in v1.2.14 

func (sp *StreamProcessor) CastToStringSafeMask(i int, val any, valType ...ColumnType) string

CastToStringSafe to masks to count bytes (even safer)

func (*StreamProcessor) CastToTime 

func (sp *StreamProcessor) CastToTime(i any) (t time.Time, err error)

CastToTime converts interface to time

func (*StreamProcessor) CastType 

func (sp *StreamProcessor) CastType(val any, typ ColumnType) any

CastType casts the type of an interface CastType is used to cast the interface place holders?

func (*StreamProcessor) CastVal 

func (sp *StreamProcessor) CastVal(i int, val any, col *Column) any

CastVal casts values with stats collection which degrades performance by ~10% go test -benchmem -run='^$ github.com/slingdata-io/sling-cli/core/dbio/iop' -bench '^BenchmarkProcessVal'

func (*StreamProcessor) CastValWithoutStats 

func (sp *StreamProcessor) CastValWithoutStats(i int, val any, typ ColumnType) any

CastValWithoutStats casts the value without counting stats

func (*StreamProcessor) CheckType added in v1.4.18 

func (sp *StreamProcessor) CheckType(v any) (typ ColumnType)

CheckType returns the type of an interface

func (*StreamProcessor) ColStats added in v1.2.10 

func (sp *StreamProcessor) ColStats() map[int]*ColumnStats

func (*StreamProcessor) CountDigits added in v1.4.20 

func (sp *StreamProcessor) CountDigits(number string) (precision, scale int)

func (*StreamProcessor) GetType 

func (sp *StreamProcessor) GetType(val any) (typ ColumnType)

GetType returns the type of an interface

func (*StreamProcessor) ParseString 

func (sp *StreamProcessor) ParseString(s string, jj ...int) (val any)

ParseString return an interface string: "varchar" integer: "integer" decimal: "decimal" date: "date" datetime: "timestamp" timestamp: "timestamp" text: "text"

func (*StreamProcessor) ParseTime 

func (sp *StreamProcessor) ParseTime(i any) (t time.Time, err error)

ParseTime parses a date string and returns time.Time

func (*StreamProcessor) ParseVal 

func (sp *StreamProcessor) ParseVal(val any) any

ParseVal parses the value into its appropriate type

func (*StreamProcessor) ProcessRow 

func (sp *StreamProcessor) ProcessRow(row []any) []any

ProcessRow processes a row

func (*StreamProcessor) ProcessVal 

func (sp *StreamProcessor) ProcessVal(val any) any

ProcessVal processes a value

func (*StreamProcessor) ResetConfig added in v1.1.15 

func (sp *StreamProcessor) ResetConfig()

func (*StreamProcessor) SetConfig 

func (sp *StreamProcessor) SetConfig(configMap map[string]string)

SetConfig sets the data.Sp.config values

func (*StreamProcessor) TruncateDecimalString added in v1.4.20 

func (sp *StreamProcessor) TruncateDecimalString(number string, decCount int) (newNumber string)

TruncateDecimalString return up to specified scale, without converting

type StringColumnTyping added in v1.4.5 

type StringColumnTyping struct {
	LengthFactor int  `json:"length_factor,omitempty" yaml:"length_factor,omitempty"`
	MinLength    int  `json:"min_length,omitempty" yaml:"min_length,omitempty"`
	MaxLength    int  `json:"max_length,omitempty" yaml:"max_length,omitempty"`
	UseMax       bool `json:"use_max,omitempty" yaml:"use_max,omitempty"`

	Note string `json:"note,omitempty" yaml:"note,omitempty"`
}

StringColumnTyping contains string type mapping configurations

func (*StringColumnTyping) Apply added in v1.4.5 

func (sct *StringColumnTyping) Apply(length, max int) (newLength int)

type TimeLevel added in v1.3.6 

type TimeLevel string
const (
	TimeLevelYear      TimeLevel = "YYYY"
	TimeLevelYearShort TimeLevel = "YY"
	TimeLevelMonthName TimeLevel = "MMM"
	TimeLevelMonth     TimeLevel = "MM"
	TimeLevelDay       TimeLevel = "DD"
	TimeLevelDayOfYear TimeLevel = "DDD"
	TimeLevelHour24    TimeLevel = "HH"
	TimeLevelHour12    TimeLevel = "hh"
	TimeLevelMinute    TimeLevel = "mm"
	TimeLevelSecond    TimeLevel = "ss"
)

func ExtractISO8601DateFields added in v1.3.6 

func ExtractISO8601DateFields(path string) (list []TimeLevel)

ExtractISO8601DateFields return a list of found date fields

func (TimeLevel) AsPartitionLevel added in v1.3.6 

func (tl TimeLevel) AsPartitionLevel() PartitionLevel

type Transform added in v1.2.2 

type Transform interface {
	Evaluate(row []any) (newRow []any, err error)
	Casted() bool
}

type TransformLegacy added in v1.4.17 

type TransformLegacy struct {
	Name       string
	Func       func(*StreamProcessor, ...any) (any, error)
	FuncString func(*StreamProcessor, string) (string, error)
	FuncTime   func(*StreamProcessor, *time.Time) error
}

type TransformLegacyList added in v1.4.17 

type TransformLegacyList []TransformLegacy

func (TransformLegacyList) HasTransform added in v1.4.17 

func (tl TransformLegacyList) HasTransform(t TransformLegacy) bool

type Transformers added in v1.2.2 

type Transformers struct {
	DecodeUTF8        transform.Transformer
	DecodeUTF8BOM     transform.Transformer
	DecodeUTF16       transform.Transformer
	DecodeISO8859_1   transform.Transformer
	DecodeISO8859_5   transform.Transformer
	DecodeISO8859_15  transform.Transformer
	DecodeWindows1250 transform.Transformer
	DecodeWindows1252 transform.Transformer

	EncodeUTF8        transform.Transformer
	EncodeUTF8BOM     transform.Transformer
	EncodeUTF16       transform.Transformer
	EncodeISO8859_1   transform.Transformer
	EncodeISO8859_5   transform.Transformer
	EncodeISO8859_15  transform.Transformer
	EncodeWindows1250 transform.Transformer
	EncodeWindows1252 transform.Transformer
}

func NewTransformers added in v1.2.2 

func NewTransformers() Transformers

type UUIDColumnTyping added in v1.6.1 

type UUIDColumnTyping struct {
	AsText bool `json:"as_text,omitempty" yaml:"as_text,omitempty"`
}

UUIDColumnTyping contains uuid type mapping configurations

func (*UUIDColumnTyping) Apply added in v1.6.1 

func (uct *UUIDColumnTyping) Apply(col *Column)

type ZStandardCompressor 

type ZStandardCompressor struct {
	Compressor
	// contains filtered or unexported fields
}

func (*ZStandardCompressor) Compress 

func (cp *ZStandardCompressor) Compress(reader io.Reader) io.Reader

Compress uses gzip to compress

func (*ZStandardCompressor) Decompress 

func (cp *ZStandardCompressor) Decompress(reader io.Reader) (sReader io.Reader, err error)

Decompress uses gzip to decompress if it is gzip. Otherwise return same reader

func (*ZStandardCompressor) Suffix 

func (cp *ZStandardCompressor) Suffix() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.