joechenrh commented on code in PR #880:
URL: https://github.com/apache/arrow-go/pull/880#discussion_r3518169144
##########
parquet/file/page_reader.go:
##########
@@ -737,6 +786,92 @@ func (p *serializedPageReader) SeekToPageWithRow(rowIdx
int64) error {
return p.err
}
+func (p *serializedPageReader) valueStream(limit io.Reader, compressed bool)
(io.Reader, io.Closer) {
+ if compressed {
+ dec := p.codec.(compress.StreamingCodec).NewReader(limit)
+ return dec, dec
+ }
+ return limit, nil
+}
+
+// drainAndClose returns the cleanup a streaming ValueBuffer runs on Close:
consume
+// any unread compressed bytes for the current page and close the decompressor.
+func drainAndClose(limit io.Reader, closer io.Closer) func() error {
+ return func() error {
+ _, _ = io.Copy(io.Discard, limit)
+ if closer != nil {
+ return closer.Close()
+ }
+ return nil
+ }
+}
+
+func (p *serializedPageReader) pageCanStream(enc format.Encoding) bool {
+ return p.columnCanStream && enc == format.Encoding_PLAIN &&
p.cryptoCtx.DataDecryptor == nil
+}
+
+func streamablePhysicalType(t parquet.Type) bool {
+ switch t {
+ case parquet.Types.ByteArray, parquet.Types.FixedLenByteArray:
+ return true
+ default:
+ return false
+ }
+}
+
+func streamableCodec(c compress.Compression) bool {
+ switch c {
+ case compress.Codecs.Uncompressed, compress.Codecs.Gzip,
compress.Codecs.Brotli, compress.Codecs.Zstd:
+ return true
+ default:
+ return false
+ }
+}
+
+// readLevelData reads the V1 rep+def level region into a buffer.
+func (p *serializedPageReader) readLevelData(r io.Reader, repEnc, defEnc
format.Encoding, numValues int) ([]byte, error) {
+ buf := make([]byte, 0, 4096)
+ readOne := func(enc format.Encoding, maxLvl int16) error {
+ switch enc {
+ case format.Encoding_RLE:
+ var hdr [4]byte
+ if _, err := io.ReadFull(r, hdr[:]); err != nil {
+ return err
+ }
+ n := int(binary.LittleEndian.Uint32(hdr[:]))
+ start := len(buf)
+ buf = append(buf, hdr[:]...)
+ buf = append(buf, make([]byte, n)...)
+ if _, err := io.ReadFull(r, buf[start+4:]); err != nil {
+ return err
+ }
+ case format.Encoding_BIT_PACKED:
+ bitWidth := bits.Len64(uint64(maxLvl))
+ n := int(bitutil.BytesForBits(int64(numValues) *
int64(bitWidth)))
Review Comment:
Nice catch, I didn't notice that my AI has modified this part during
reviewing. Updated in
https://github.com/apache/arrow-go/pull/880/changes/aeeb9920fbbcbb6164b8af6175e9beaae3b9935c
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
