package parser import ( "bufio" "fmt" "io" "os" "sync" "github.com/tidwall/gjson" ) // countingReader wraps an io.Reader and counts bytes read. type countingReader struct { r io.Reader n int64 } func (cr *countingReader) Read(p []byte) (int, error) { n, err := cr.r.Read(p) cr.n += int64(n) return n, err } // lineReader reads JSONL files line by line, skipping lines that // exceed maxLen rather than aborting. The buffer starts small and // grows on demand up to maxLen. After iteration, call Err() to // check for I/O errors (as opposed to normal EOF). type lineReader struct { r *bufio.Reader cr *countingReader maxLen int buf []byte err error bytesRead int64 // total bytes consumed (from countingReader) } const maxPooledLineBufferSize = 256 << 10 var lineReaderPool = sync.Pool{ New: func() any { cr := new(countingReader) return &lineReader{ r: bufio.NewReaderSize(cr, initialScanBufSize), cr: cr, } }, } func newLineReader(r io.Reader, maxLen int) *lineReader { lr := lineReaderPool.Get().(*lineReader) lr.cr.r = r lr.cr.n = 0 lr.r.Reset(lr.cr) lr.maxLen = maxLen lr.buf = lr.buf[:0] lr.err = nil lr.bytesRead = 0 return lr } func releaseLineReader(lr *lineReader) { // Do not let an exceptional long line pin a multi-megabyte backing // array. sync.Pool may discard the remaining workspace at any GC. if cap(lr.buf) > maxPooledLineBufferSize { lr.buf = nil } else { lr.buf = lr.buf[:0] } lr.r.Reset(nil) lr.cr.r = nil lr.cr.n = 0 lr.maxLen = 0 lr.err = nil lr.bytesRead = 0 lineReaderPool.Put(lr) } // next returns the next line (without trailing newline) and true, // or ("", false) at EOF or read error. After the loop, call Err() // to distinguish EOF from I/O failure. func (lr *lineReader) next() (string, bool) { for { line, err := lr.readLine() if err != nil { if err != io.EOF { lr.err = err } return "", false } if line != "" { return line, true } // Empty line or skipped oversized line — continue. } } // Err returns the first non-EOF read error encountered, or nil. func (lr *lineReader) Err() error { return lr.err } // readLine reads a full line, returning "" for blank/oversized // lines and a non-nil error only at EOF or read failure. // updateBytesRead computes total bytes consumed by // subtracting any buffered-but-not-consumed data from the // countingReader total. func (lr *lineReader) updateBytesRead() { if lr.cr != nil { lr.bytesRead = lr.cr.n - int64(lr.r.Buffered()) } } func (lr *lineReader) readLine() (string, error) { lr.buf = lr.buf[:0] oversized := false for { chunk, isPrefix, err := lr.r.ReadLine() if err != nil { if len(lr.buf) > 0 && err == io.EOF { lr.updateBytesRead() break } return "", err } if oversized { if !isPrefix { lr.updateBytesRead() return "", nil // done skipping } continue } // ReadLine's common case is a complete line backed by the // bufio.Reader buffer. Convert it directly instead of allocating // a second initialScanBufSize scratch buffer for every file. if len(lr.buf) == 0 && !isPrefix { lr.updateBytesRead() if len(chunk) > lr.maxLen { return "", nil } return string(chunk), nil } lr.buf = append(lr.buf, chunk...) if len(lr.buf) > lr.maxLen { oversized = true lr.buf = lr.buf[:0] if !isPrefix { lr.updateBytesRead() return "", nil } continue } if !isPrefix { lr.updateBytesRead() break } } return string(lr.buf), nil } // readJSONLFrom opens a JSONL file, seeks to offset, and // calls fn for each valid JSON line. Returns the number of // bytes consumed (relative to offset) and any I/O error. The // returned byte count covers only complete lines, so callers // can use offset+consumed as a safe resume point even if the // file had a partially written line at EOF. func readJSONLFrom( path string, offset int64, fn func(line string), ) (consumed int64, err error) { f, err := os.Open(path) if err != nil { return 0, fmt.Errorf("open %s: %w", path, err) } defer f.Close() if _, err := f.Seek(offset, io.SeekStart); err != nil { return 0, fmt.Errorf( "seek %s to %d: %w", path, offset, err, ) } lr := newLineReader(f, maxLineSize) defer releaseLineReader(lr) for { line, ok := lr.next() if !ok { break } if gjson.Valid(line) { fn(line) // Track offset through last valid JSON line // so partial lines at EOF are not skipped. consumed = lr.bytesRead } } return consumed, lr.Err() }