iop

package

Go to main page

Versions in this module

v1

v1.2.10

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

v1.2.9

Apr 28, 2024

v1.2.8

Apr 28, 2024

v1.2.7

Apr 28, 2024

Changes in this version

type Iterator

+ StreamRowNum uint64

type StreamConfig

+ NullAs string

v1.2.6

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

v1.2.5

Apr 15, 2024

v1.2.4

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

v1.2.3

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

v1.2.2

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

v1.2.1

Mar 23, 2024

v1.2.0

Mar 23, 2024

v1.1.25

Mar 23, 2024

v1.1.24

Mar 23, 2024

v1.1.23

Mar 23, 2024

v1.1.22

Mar 23, 2024

v1.1.21

Mar 23, 2024

v1.1.20

Mar 23, 2024

v1.1.19

Mar 23, 2024

v1.1.18

Mar 23, 2024

v1.1.17

Mar 23, 2024

v1.1.16

Mar 23, 2024

v1.1.15

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()

v1.1.14

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)

v1.1.13

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

v1.1.12

Mar 1, 2024

Changes in this version

type Column

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

v1.1.11

Feb 28, 2024

v1.1.10

Feb 28, 2024

v1.1.9

Feb 28, 2024

v1.1.8

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)

v1.1.7

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

v1.1.6

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)

v1.1.5

Feb 11, 2024

v1.1.4

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

v1.1.3

Feb 8, 2024

v1.1.2

Feb 8, 2024

v1.1.1

Feb 7, 2024

v1.1.0

Feb 7, 2024

v1.0.74

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

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