iop

Feb 1, 2025

Changes in this version

type HttpStreamPart

+ Columns Columns

Jan 22, 2025

Changes in this version

+ var TimeLevelAscending = []TimeLevel

+ var TimeLevelDescending = []TimeLevel

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

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

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

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

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

type DuckDb

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

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

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

+ type HttpStreamPart struct

+ FromExpr string

+ Index int

type PartitionLevel

+ const PartitionLevelSecond

+ func GetLowestPartTimeLevel(url string) (PartitionLevel, error)

+ func (level PartitionLevel) Str() string

+ type TimeLevel string

+ const TimeLevelDay

+ const TimeLevelDayOfYear

+ const TimeLevelHour12

+ const TimeLevelHour24

+ const TimeLevelMinute

+ const TimeLevelMonth

+ const TimeLevelMonthName

+ const TimeLevelSecond

+ const TimeLevelYear

+ const TimeLevelYearShort

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

+ func (tl TimeLevel) AsPartitionLevel() PartitionLevel

Jan 15, 2025

Changes in this version

+ var DuckDbURISeparator = "|-|+|"

type ColumnType

+ const UUIDType

type FileStreamConfig

+ DuckDBFilename bool

+ func (sc *FileStreamConfig) ComputeWithDuckDB() bool

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

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

Dec 27, 2024

Changes in this version

+ var PartitionLevelsAscending = []PartitionLevel

+ var PartitionLevelsDescending = []PartitionLevel

+ var TransformLower = Transform

+ var TransformUpper = Transform

type ColumnStats

+ LastVal any

type CompressorType

+ func (ct CompressorType) Normalize() *CompressorType

+ func (ct CompressorType) String() string

type DuckDbCopyOptions

+ Format dbio.FileType

+ type PartitionLevel string

+ const PartitionLevelDay

+ const PartitionLevelHour

+ const PartitionLevelMinute

+ const PartitionLevelMonth

+ const PartitionLevelWeek

+ const PartitionLevelYear

+ const PartitionLevelYearMonth

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

+ func (level PartitionLevel) IsValid() bool

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

Dec 1, 2024

Changes in this version

+ var TransformEmptyAsNull = Transform

type Dataflow

+ func (df *Dataflow) BufferDataset() Dataset

type DuckDb

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

+ type DuckDbCopyOptions struct

+ Compression string

+ FileSizeBytes int64

+ PartitionFields []string

+ PartitionKey string

+ WritePartitionCols bool

+ type TransformList []Transform

+ func (tl TransformList) HasTransform(t Transform) bool

Nov 24, 2024

Changes in this version

type Metadata

+ ExecID KeyValue

Oct 24, 2024

Changes in this version

+ type ColumnCasing string

+ const LowerColumnCasing

+ const SnakeColumnCasing

+ const SourceColumnCasing

+ const TargetColumnCasing

+ const UpperColumnCasing

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

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

+ func (cc *ColumnCasing) IsEmpty() bool

type StreamConfig

+ ColumnCasing ColumnCasing

+ FileMaxBytes int64

+ Sheet string

+ func LoaderStreamConfig(header bool) StreamConfig

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

Oct 21, 2024

Changes in this version

+ func FormatValue(val any, column Column, connType dbio.Type) (newVal string)

type CSV

+ Quote string

type FileStreamConfig

+ FileSelect *[]string

type StreamConfig

+ Quote string

Sep 22, 2024

Changes in this version

type ColumnType

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

+ type CsvDuckDb struct

+ Duck *DuckDb

+ URI string

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

+ func (r *CsvDuckDb) Close() error

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

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

type Datastream

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

type DeltaReader

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

type DuckDb

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

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

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

+ type FileStreamConfig struct

+ Format dbio.FileType

+ IncrementalKey string

+ IncrementalValue string

+ Limit int

+ Props map[string]string

+ SQL string

+ Select []string

+ func (sc *FileStreamConfig) ShouldUseDuckDB() bool

type IcebergReader

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

type ParquetDuckDb

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

Aug 27, 2024

Changes in this version

+ var DuckDbUseTempFile = false

+ var DuckDbVersion = "1.0.0"

+ var TransformSetTimezone = Transform

+ var TransformsMap = map[string]Transform

+ func EnsureBinDuckDB(version string) (binPath string, err error)

type Column

+ Constraint *ColumnConstraint

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

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

+ func (col *Column) SetConstraint()

+ type ColumnConstraint struct

+ Errors []string

+ EvalFunc ConstraintEvalFunc

+ Expression string

+ FailCnt uint64

type ColumnType

+ func DuckDBTypeToColumnType(duckDBType string) ColumnType

+ type ConstraintEvalFunc func(value any) bool

type Datastream

+ func (ds *Datastream) ConsumeDeltaReader(uri string, fields []string, limit uint64, fsProps map[string]string) (err error)

+ func (ds *Datastream) ConsumeIcebergReader(uri string, fields []string, limit uint64, fsProps map[string]string) (err error)

+ func (ds *Datastream) ConsumeParquetReaderDuckDb(uri string, fields []string, limit uint64, fsProps map[string]string) (err error)

+ type DeltaReader struct

+ Duck *DuckDb

+ URI string

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

+ func (d *DeltaReader) Close() error

+ func (d *DeltaReader) Columns() (Columns, error)

+ func (d *DeltaReader) MakeSelectQuery(fields []string, limit uint64) string

+ type DuckDb struct

+ Context *g.Context

+ Proc *process.Proc

+ StdinWriter io.Writer

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

+ func (duck *DuckDb) AddExtension(extension string)

+ func (duck *DuckDb) Close() error

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

+ func (duck *DuckDb) Exec(sql string, args ...interface{}) (result sql.Result, err error)

+ func (duck *DuckDb) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error)

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

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

+ func (duck *DuckDb) Import(tableFName string, ds *Datastream) (count uint64, err error)

+ func (duck *DuckDb) MakeScanSelectQuery(scanFunc, uri string, fields []string, limit uint64) (sql string)

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

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

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

+ func (duck *DuckDb) Query(sql string, options ...map[string]interface{}) (data Dataset, err error)

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

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

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

+ func (duck *DuckDb) Stream(sql string, options ...map[string]interface{}) (ds *Datastream, err error)

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

+ type IcebergReader struct

+ Duck *DuckDb

+ URI string

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

+ func (i *IcebergReader) Close() error

+ func (i *IcebergReader) Columns() (Columns, error)

+ func (i *IcebergReader) MakeSelectQuery(fields []string, limit uint64) string

+ type ParquetDuckDb struct

+ Duck *DuckDb

+ URI string

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

+ func (p *ParquetDuckDb) Close() error

+ func (p *ParquetDuckDb) Columns() (Columns, error)

+ func (p *ParquetDuckDb) MakeSelectQuery(fields []string, limit uint64) string

type Transform

+ FuncString func(*StreamProcessor, string) (string, error)

+ FuncTime func(*StreamProcessor, *time.Time) error

+ Name string

Aug 14, 2024

Changes in this version

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

type Columns

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

type Dataflow

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

type KeyType

+ func (kt KeyType) MetadataKey() string

Jul 25, 2024

Changes in this version

type Datastream

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

type StreamProcessor

+ func (sp *StreamProcessor) CastToStringSafeMask(i int, val interface{}, valType ...ColumnType) string

Jun 29, 2024

Changes in this version

type ParquetWriter

+ WriterMap *parquet.GenericWriter[map[string]any]

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

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

type StreamProcessor

+ func (sp *StreamProcessor) CastToBool(i interface{}) (b bool, err error)

type Transform

+ const TransformParseMsUuid

Jun 4, 2024

Changes in this version

type Dataflow

+ func (df *Dataflow) SetBatchLimit(limit int64)

type StreamConfig

+ BatchLimit int64

May 18, 2024

Changes in this version

+ var AllCompressorType = []struct{ ... }

type ColumnStats

+ DateTimeZCnt int64

type Iterator

+ func (it *Iterator) BelowEqualIncrementalVal() bool

type StreamProcessor

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

Apr 28, 2024

Changes in this version

type Iterator

+ StreamRowNum uint64

type StreamConfig

+ NullAs string

Apr 22, 2024

Changes in this version

type Datastream

+ func (ds *Datastream) CastRowToStringSafe(row []any) []string

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

+ type ReaderReady struct

+ Reader io.Reader

+ URI string

type StreamProcessor

+ func (sp *StreamProcessor) CastToStringSafe(i int, val interface{}, valType ...ColumnType) string

Apr 13, 2024

Changes in this version

type CSV

+ Escape string

type Datastream

+ func (ds *Datastream) ConsumeCsvReaderChl(readerChn chan io.Reader) (err error)

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

type StreamConfig

+ Escape string

Apr 5, 2024

Changes in this version

type Column

+ func (col *Column) IsBinary() bool

type ColumnType

+ func (ct ColumnType) IsBinary() bool

type StreamProcessor

+ func (sp *StreamProcessor) EncodingTransform(t Transform, val string) (newVal string, err error)

type Transform

+ const TransformEncodeLatin1

+ const TransformEncodeLatin5

+ const TransformEncodeLatin9

+ const TransformEncodeUtf16

+ const TransformEncodeUtf8

+ const TransformEncodeUtf8Bom

+ const TransformEncodeWindows1250

+ const TransformEncodeWindows1252

type Transformers

+ DecodeISO8859_1 transform.Transformer

+ DecodeISO8859_15 transform.Transformer

+ DecodeISO8859_5 transform.Transformer

+ DecodeUTF16 transform.Transformer

+ DecodeUTF8 transform.Transformer

+ DecodeUTF8BOM transform.Transformer

+ DecodeWindows1250 transform.Transformer

+ DecodeWindows1252 transform.Transformer

+ EncodeISO8859_1 transform.Transformer

+ EncodeISO8859_15 transform.Transformer

+ EncodeISO8859_5 transform.Transformer

+ EncodeUTF16 transform.Transformer

+ EncodeUTF8 transform.Transformer

+ EncodeUTF8BOM transform.Transformer

+ EncodeWindows1250 transform.Transformer

+ EncodeWindows1252 transform.Transformer

Mar 31, 2024

Changes in this version

type Columns

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

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

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

type Dataflow

+ EgressBytes uint64

+ func (df *Dataflow) AddEgressBytes(bytes uint64)

type Dataset

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

type Datastream

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

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

+ func (ds *Datastream) NewExcelReaderChnl(rowLimit int, bytesLimit int64, sheetName string) (readerChn chan *BatchReader)

+ type Excel struct

+ File *excelize.File

+ Path string

+ Sheets []string

+ func NewExcel() (xls *Excel)

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

+ func NewExcelFromReader(reader io.Reader) (xls *Excel, err error)

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

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

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

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

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

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

+ type GoogleSheet struct

+ Sheets []string

+ SpreadsheetID string

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

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

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

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

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

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

+ func (ggs *GoogleSheet) URL() string

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

type ParquetArrowReader

+ Context *g.Context

type StreamConfig

+ Map map[string]string

type StreamProcessor

+ func (sp *StreamProcessor) DecodeValue(t Transform, val string) (newVal string, err error)

+ type Transform string

+ const TransformDecodeLatin1

+ const TransformDecodeLatin5

+ const TransformDecodeLatin9

+ const TransformDecodeUtf16

+ const TransformDecodeUtf8

+ const TransformDecodeUtf8Bom

+ const TransformDecodeWindows1250

+ const TransformDecodeWindows1252

+ const TransformDuckdbListToText

+ const TransformHashMd5

+ const TransformHashSha256

+ const TransformHashSha512

+ const TransformParseBit

+ const TransformParseFix

+ const TransformParseUuid

+ const TransformReplace0x00

+ const TransformReplaceAccents

+ const TransformReplaceNonPrintable

+ const TransformTrimSpace

+ type Transformers struct

+ Accent transform.Transformer

+ ISO8859_1 transform.Transformer

+ ISO8859_15 transform.Transformer

+ ISO8859_5 transform.Transformer

+ UTF16 transform.Transformer

+ UTF8 transform.Transformer

+ UTF8BOM transform.Transformer

+ Windows1250 transform.Transformer

+ Windows1252 transform.Transformer

+ func NewTransformers() Transformers

Mar 23, 2024

Changes in this version

type CSV

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

type Column

+ FileURI string

type Columns

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

type Dataflow

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

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

+ func (df *Dataflow) ResetConfig()

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

type Datastream

+ func (ds *Datastream) SetFileURI()

+ type StreamConfig struct

+ BoolAsInt bool

+ Columns Columns

+ Compression string

+ DatetimeFormat string

+ Delimiter string

+ EmptyAsNull bool

+ FieldsPerRec int

+ FileMaxRows int64

+ Flatten bool

+ Header bool

+ Jmespath string

+ MaxDecimals int

+ NullIf string

+ SkipBlankLines bool

+ TrimSpace bool

+ func DefaultStreamConfig() *StreamConfig

type StreamProcessor

+ Config *StreamConfig

+ func (sp *StreamProcessor) ResetConfig()

Mar 14, 2024

Changes in this version

type Column

+ func (col *Column) IsDate() bool

+ func (col *Column) IsFloat() bool

type ColumnStats

+ DateTimeCnt int64

type ColumnType

+ func (ct ColumnType) IsFloat() bool

type Datastream

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

Mar 7, 2024

Changes in this version

+ func ParseFIX(message string) (fixMap map[string]any, err error)

+ func ReplaceNonPrintable(val string) string

type Iterator

+ RowIsCasted bool

Mar 1, 2024

Changes in this version

type Column

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

Feb 26, 2024

Changes in this version

type ColumnType

+ func (ct ColumnType) IsDate() bool

type Columns

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

+ func (cols Columns) PrettyTable() (output string)

type Dataset

+ func (data *Dataset) PrettyTable() (output string)

Feb 22, 2024

Changes in this version

type Column

+ func (col *Column) HasNulls() bool

+ func (col *Column) HasNullsPlus1() bool

type Datastream

+ func (ds *Datastream) NewParquetArrowReaderChnl(rowLimit int, bytesLimit int64, compression CompressorType) (readerChn chan *BatchReader)

type Parquet

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

type ParquetArrowReader

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

+ type ParquetWriter struct

+ Writer *parquet.Writer

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

+ func (pw *ParquetWriter) Close() error

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

Feb 19, 2024

Changes in this version

+ func DecimalByteArrayToString(dec []byte, precision int, scale int) string

+ func MakeDecNumScale(scale int) *big.Rat

+ func StrIntToBinary(num string, order string, length int, signed bool) string

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

type Columns

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

+ type ParquetArrowDumper struct

+ func NewParquetArrowDumper(ccReader file.ColumnChunkReader) *ParquetArrowDumper

+ func (pad *ParquetArrowDumper) Next() (interface{}, bool)

+ type ParquetArrowReader struct

+ Data *Dataset

+ Path string

+ Reader *file.Reader

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

+ func (p *ParquetArrowReader) Columns() Columns

+ type ParquetArrowWriter struct

+ Writer *file.Writer

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

+ func (p *ParquetArrowWriter) AppendNewRowGroup() (err error)

+ func (p *ParquetArrowWriter) Close() (err error)

+ func (p *ParquetArrowWriter) Columns() Columns

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

Feb 11, 2024

Changes in this version

+ var KeyTypes = []KeyType

type KeyType

+ const ClusterKey

+ const DistributionKey

+ const DuplicateKey

+ const HashKey

+ const IndexKey

+ const PartitionKey

+ const UniqueKey

type Metadata

+ RowID KeyValue

+ RowNum KeyValue

Feb 7, 2024

Changes in this version

+ var RemoveTrailingDecZeros = false

+ var SampleSize = 900

+ var Transforms = map[string]TransformFunc

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

+ func CleanHeaderRow(header []string) []string

+ func CleanName(name string) (newName string)

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

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

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

+ func IsDummy(columns []Column) bool

+ func Iso8601ToGoLayout(dateFormat string) (goDateFormat string)

+ func MakeRowsChan() chan []any

+ func NewJSONStream(ds *Datastream, decoder decoderLike, flatten bool, jmespath string) *jsonStream

+ func ReplaceAccents(sp *StreamProcessor, val string) (string, error)

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

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

+ func Unzip(src string, dest string) ([]string, error)

+ type Avro struct

+ Data *Dataset

+ Path string

+ Reader *goavro.OCFReader

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

+ func (a *Avro) Columns() Columns

+ type Batch struct

+ Columns Columns

+ Count int64

+ Limit int64

+ Previous *Batch

+ Rows chan []any

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

+ func (b *Batch) Close()

+ func (b *Batch) ColumnsChanged() bool

+ func (b *Batch) Ds() *Datastream

+ func (b *Batch) ID() string

+ func (b *Batch) IsFirst() bool

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

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

+ type BatchReader struct

+ Batch *Batch

+ Columns Columns

+ Counter int

+ Reader io.Reader

+ type CSV struct

+ Columns []Column

+ Config map[string]string

+ Data Dataset

+ Delimiter rune

+ FieldsPerRecord int

+ File *os.File

+ NoDebug bool

+ NoHeader bool

+ Path string

+ Reader io.Reader

+ func (c *CSV) InferSchema() error

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

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

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

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

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

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

+ type Column struct

+ Database string

+ DbPrecision int

+ DbScale int

+ DbType string

+ Description string

+ Metadata map[string]string

+ Name string

+ Position int

+ Schema string

+ Sourced bool

+ Stats ColumnStats

+ Table string

+ Type ColumnType

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

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

+ func (col *Column) IsBool() bool

+ func (col *Column) IsDatetime() bool

+ func (col *Column) IsDecimal() bool

+ func (col *Column) IsInteger() bool

+ func (col *Column) IsNumber() bool

+ func (col *Column) IsString() bool

+ func (col *Column) IsUnique() bool

+ func (col *Column) Key() string

+ func (col *Column) SetLengthPrecisionScale()

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

+ type ColumnStats struct

+ BoolCnt int64

+ Checksum uint64

+ DateCnt int64

+ DecCnt int64

+ IntCnt int64

+ JsonCnt int64

+ Max int64

+ MaxDecLen int

+ MaxLen int

+ Min int64

+ MinLen int

+ NullCnt int64

+ StringCnt int64

+ TotalCnt int64

+ UniqCnt int64

+ func (cs *ColumnStats) DistinctPercent() float64

+ func (cs *ColumnStats) DuplicateCount() int64

+ func (cs *ColumnStats) DuplicatePercent() float64

+ type ColumnType string

+ const BigIntType

+ const BinaryType

+ const BoolType

+ const DateType

+ const DatetimeType

+ const DecimalType

+ const FloatType

+ const IntegerType

+ const JsonType

+ const SmallIntType

+ const StringType

+ const TextType

+ const TimeType

+ const TimestampType

+ const TimestampzType

+ const TimezType

+ func (ct ColumnType) IsBool() bool

+ func (ct ColumnType) IsDatetime() bool

+ func (ct ColumnType) IsDecimal() bool

+ func (ct ColumnType) IsInteger() bool

+ func (ct ColumnType) IsJSON() bool

+ func (ct ColumnType) IsNumber() bool

+ func (ct ColumnType) IsString() bool

+ func (ct ColumnType) IsValid() bool

+ type Columns []Column

+ func NewColumns(cols ...Column) Columns

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

+ func (cols Columns) Add(newCols Columns, overwrite bool) (col2 Columns, added Columns)

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

+ func (cols Columns) Coerce(castCols Columns, hasHeader bool) (newCols Columns)

+ func (cols Columns) Dataset() Dataset

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

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

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

+ func (cols Columns) GetKeys(keyType KeyType) Columns

+ func (cols Columns) IsDifferent(newCols Columns) bool

+ func (cols Columns) IsDummy() bool

+ func (cols Columns) IsSimilarTo(otherCols Columns) bool

+ func (cols Columns) Keys() []string

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

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

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

+ func (cols Columns) SetKeys(keyType KeyType, names ...string)

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

+ type Compressor interface

+ Compress func(io.Reader) io.Reader

+ Decompress func(io.Reader) (io.Reader, error)

+ Self func() Compressor

+ Suffix func() string

+ func NewCompressor(cpType CompressorType) Compressor

+ type CompressorType string

+ const AutoCompressorType

+ const GzipCompressorType

+ const NoneCompressorType

+ const SnappyCompressorType

+ const ZStandardCompressorType

+ const ZipCompressorType

+ func CompressorTypePtr(v CompressorType) *CompressorType

+ type Dataflow struct

+ Buffer [][]interface{}

+ Columns Columns

+ Context *g.Context

+ FsURL string

+ InBytes uint64

+ Inferred bool

+ Limit uint64

+ OnColumnAdded func(col Column) error

+ OnColumnChanged func(col Column) error

+ OutBytes uint64

+ Ready bool

+ SchemaVersion int

+ StreamCh chan *Datastream

+ StreamMap map[string]*Datastream

+ Streams []*Datastream

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

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

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

+ func (df *Dataflow) AddInBytes(bytes uint64)

+ func (df *Dataflow) AddOutBytes(bytes uint64)

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

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

+ func (df *Dataflow) CleanUp()

+ func (df *Dataflow) Close()

+ func (df *Dataflow) CloseCurrentBatches()

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

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

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

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

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

+ func (df *Dataflow) IsClosed() bool

+ func (df *Dataflow) IsEmpty() bool

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

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

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

+ func (df *Dataflow) SetColumns(columns []Column)

+ func (df *Dataflow) SetEmpty()

+ func (df *Dataflow) SetReady()

+ func (df *Dataflow) Size() int

+ func (df *Dataflow) SyncColumns()

+ func (df *Dataflow) SyncStats()

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

+ func (df *Dataflow) WaitClosed()

+ func (df *Dataflow) WaitReady() error

+ type Dataset struct

+ Columns Columns

+ Duration float64

+ Inferred bool

+ NoDebug bool

+ Result *sqlx.Rows

+ Rows [][]interface{}

+ SQL string

+ SafeInference bool

+ Sp *StreamProcessor

+ func NewDataset(columns Columns) (data Dataset)

+ func NewDatasetFromMap(m map[string]interface{}) (data Dataset)

+ func ReadCsv(path string) (Dataset, error)

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

+ func (data *Dataset) Append(row []interface{})

+ func (data *Dataset) ColValues(col int) []interface{}

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

+ func (data *Dataset) FirstRow() []interface{}

+ func (data *Dataset) FirstVal() interface{}

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

+ func (data *Dataset) InferColumnTypes()

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

+ func (data *Dataset) Print(limit int)

+ func (data *Dataset) Records(lower ...bool) []map[string]interface{}

+ func (data *Dataset) RecordsCasted(lower ...bool) []map[string]interface{}

+ func (data *Dataset) RecordsString(lower ...bool) []map[string]interface{}

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

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

+ func (data *Dataset) Stream() *Datastream

+ func (data *Dataset) ToJSONMap() map[string]interface{}

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

+ type Datastream struct

+ BatchChan chan *Batch

+ Batches []*Batch

+ Buffer [][]any

+ Bytes uint64

+ Columns Columns

+ Context *g.Context

+ Count uint64

+ CurrentBatch *Batch

+ ID string

+ Inferred bool

+ Metadata Metadata

+ NoDebug bool

+ Ready bool

+ SafeInference bool

+ Sp *StreamProcessor

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

+ func NewDatastream(columns Columns) (ds *Datastream)

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

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

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

+ func (ds *Datastream) AddBytes(b int64)

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

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

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

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

+ func (ds *Datastream) Close()

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

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

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

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

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

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

+ func (ds *Datastream) ConsumeParquetReaderSeeker(reader io.ReaderAt) (err error)

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

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

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

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

+ func (ds *Datastream) Df() *Dataflow

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

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

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

+ func (ds *Datastream) IsClosed() bool

+ func (ds *Datastream) LatestBatch() *Batch

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

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

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

+ func (ds *Datastream) NewCsvBufferReader(limit int, bytesLimit int64) *bytes.Reader

+ func (ds *Datastream) NewCsvBufferReaderChnl(rowLimit int, bytesLimit int64) (readerChn chan *bytes.Reader)

+ func (ds *Datastream) NewCsvBytesChnl(chunkRowSize int) (dataChn chan *[]byte)

+ func (ds *Datastream) NewCsvReader(rowLimit int, bytesLimit int64) *io.PipeReader

+ func (ds *Datastream) NewCsvReaderChnl(rowLimit int, bytesLimit int64) (readerChn chan *BatchReader)

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

+ func (ds *Datastream) NewJsonLinesReaderChnl(rowLimit int, bytesLimit int64) (readerChn chan *io.PipeReader)

+ func (ds *Datastream) NewJsonReaderChnl(rowLimit int, bytesLimit int64) (readerChn chan *io.PipeReader)

+ func (ds *Datastream) NewParquetReaderChnl(rowLimit int, bytesLimit int64, compression CompressorType) (readerChn chan *BatchReader)

+ func (ds *Datastream) Pause()

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

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

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

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

+ func (ds *Datastream) SetEmpty()

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

+ func (ds *Datastream) SetMetadata(jsonStr string)

+ func (ds *Datastream) SetReady()

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

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

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

+ func (ds *Datastream) TryPause() bool

+ func (ds *Datastream) Unpause()

+ func (ds *Datastream) WaitClosed()

+ func (ds *Datastream) WaitReady() error

+ type GzipCompressor struct

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

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

+ func (cp *GzipCompressor) Suffix() string

+ type Iterator struct

+ Closed bool

+ Context *g.Context

+ Counter uint64

+ IsCasted bool

+ Reprocess chan []any

+ Row []any

+ func (it *Iterator) Ds() *Datastream

+ type KeyType string

+ const AggregateKey

+ const PrimaryKey

+ const SortKey

+ const UpdateKey

+ type KeyValue struct

+ Key string

+ Value any

+ type Metadata struct

+ LoadedAt KeyValue

+ StreamURL KeyValue

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

+ type NoneCompressor struct

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

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

+ func (cp *NoneCompressor) Suffix() string

+ type Parquet struct

+ Data *Dataset

+ Path string

+ Reader *parquet.Reader

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

+ func (p *Parquet) Columns() Columns

+ type RecNode struct

+ func NewRecNode(cols Columns) *RecNode

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

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

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

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

+ func (rn *RecNode) ID() int

+ func (rn *RecNode) Leaf() bool

+ func (rn *RecNode) Optional() bool

+ func (rn *RecNode) Repeated() bool

+ func (rn *RecNode) Required() bool

+ func (rn *RecNode) String() string

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

+ type Record struct

+ Columns *Columns

+ Values []any

+ type SAS struct

+ Data *Dataset

+ Path string

+ Reader *datareader.SAS7BDAT

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

+ func (s *SAS) Columns() Columns

+ type SSHClient struct

+ Err error

+ Host string

+ Passphrase string

+ Password string

+ Port int

+ PrivateKey string

+ TgtHost string

+ TgtPort int

+ User string

+ func (s *SSHClient) Close()

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

+ func (s *SSHClient) GetOutput() (stdout string, stderr string)

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

+ func (s *SSHClient) RunAsProcess() (localPort int, err error)

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

+ type Shaper struct

+ ColMap map[int]int

+ Func func([]any) []any

+ SrcColumns Columns

+ TgtColumns Columns

+ type SnappyCompressor struct

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

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

+ func (cp *SnappyCompressor) Suffix() string

+ type StreamProcessor struct

+ N uint64

+ func NewStreamProcessor() *StreamProcessor

+ func (sp *StreamProcessor) CastRow(row []interface{}, columns Columns) []interface{}

+ func (sp *StreamProcessor) CastToString(i int, val interface{}, valType ...ColumnType) string

+ func (sp *StreamProcessor) CastToTime(i interface{}) (t time.Time, err error)

+ func (sp *StreamProcessor) CastType(val interface{}, typ ColumnType) interface{}

+ func (sp *StreamProcessor) CastVal(i int, val interface{}, col *Column) interface{}

+ func (sp *StreamProcessor) CastValWithoutStats(i int, val interface{}, typ ColumnType) interface{}

+ func (sp *StreamProcessor) GetType(val interface{}) (typ ColumnType)

+ func (sp *StreamProcessor) ParseString(s string, jj ...int) interface{}

+ func (sp *StreamProcessor) ParseTime(i interface{}) (t time.Time, err error)

+ func (sp *StreamProcessor) ParseVal(val interface{}) interface{}

+ func (sp *StreamProcessor) ProcessRow(row []interface{}) []interface{}

+ func (sp *StreamProcessor) ProcessVal(val interface{}) interface{}

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

+ type TransformFunc func(*StreamProcessor, string) (string, error)

+ type ZStandardCompressor struct

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

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

+ func (cp *ZStandardCompressor) Suffix() string