zeroshade commented on code in PR #911:
URL: https://github.com/apache/arrow-go/pull/911#discussion_r3554173550
##########
parquet/file/page_reader.go:
##########
@@ -518,53 +518,85 @@ func (p *serializedPageReader) decompress(rd io.Reader,
lenCompressed int, buf [
return p.codec.Decode(buf, data), nil
}
-func (p *serializedPageReader) readV2Encrypted(rd io.Reader, lenCompressed
int, levelsBytelen int, compressed bool, buf []byte) error {
+func (p *serializedPageReader) readV2Encrypted(rd io.Reader, lenCompressed
int, lenUncompressed int, levelsBytelen int, compressed bool, buf []byte) (int,
error) {
// if encrypted, we need to decrypt before decompressing
p.decompressBuffer.ResizeNoShrink(lenCompressed)
n, err := io.ReadFull(rd, p.decompressBuffer.Bytes()[:lenCompressed])
if err != nil {
- return err
+ return n, err
}
if n != lenCompressed {
- return fmt.Errorf("parquet: expected to read %d compressed
bytes, got %d", lenCompressed, n)
+ return n, fmt.Errorf("parquet: expected to read %d compressed
bytes, got %d", lenCompressed, n)
}
data :=
p.cryptoCtx.DataDecryptor.Decrypt(p.decompressBuffer.Bytes()[:lenCompressed])
+
// encrypted + uncompressed -> just copy the decrypted data to output
buffer
if !compressed {
- copy(buf, data)
- return nil
+ if len(data) != lenUncompressed {
+ return n, fmt.Errorf("parquet: metadata said %d bytes
uncompressed data page, got %d bytes", lenUncompressed, len(data))
+ }
+
+ copied := copy(buf, data)
+ if copied != lenUncompressed {
+ return copied, fmt.Errorf("parquet: expected to read %d
uncompressed encrypted bytes, got %d", lenUncompressed, copied)
+ }
+ return copied, nil
}
// definition + repetition levels are always uncompressed
if levelsBytelen > 0 {
- copy(buf, data[:levelsBytelen])
+ if len(data) < levelsBytelen {
+ return n, fmt.Errorf("parquet: expected to read %d
bytes for levels, got %d", levelsBytelen, len(data))
+ }
+ copied := copy(buf, data[:levelsBytelen])
+ if copied != levelsBytelen {
+ return copied, fmt.Errorf("parquet: expected to read %d
bytes for levels, got %d", levelsBytelen, copied)
+ }
data = data[levelsBytelen:]
}
- p.codec.Decode(buf[levelsBytelen:], data)
- return nil
+ decoded := p.codec.Decode(buf[levelsBytelen:], data)
+ if len(decoded) != lenUncompressed-levelsBytelen {
+ return levelsBytelen + len(decoded), fmt.Errorf("parquet:
metadata said %d bytes uncompressed data page, got %d bytes", lenUncompressed,
levelsBytelen+len(decoded))
+ }
+ return levelsBytelen + len(decoded), nil
}
-func (p *serializedPageReader) readV2Unencrypted(rd io.Reader, lenCompressed
int, levelsBytelen int, compressed bool, buf []byte) error {
+func (p *serializedPageReader) readV2Unencrypted(rd io.Reader, lenCompressed
int, lenUncompressed int, levelsBytelen int, compressed bool, buf []byte) (int,
error) {
if !compressed {
// uncompressed, just read into the buffer
- if _, err := io.ReadFull(rd, buf); err != nil {
- return err
+ n, err := io.ReadFull(rd, buf)
+ if err != nil {
+ return n, err
+ }
+ if n != lenUncompressed {
+ return n, fmt.Errorf("parquet: expected to read %d
uncompressed bytes, got %d", lenUncompressed, n)
}
- return nil
+ return n, nil
}
+ decodedLen := 0
// definition + repetition levels are always uncompressed
if levelsBytelen > 0 {
- if _, err := io.ReadFull(rd, buf[:levelsBytelen]); err != nil {
- return err
+ n, err := io.ReadFull(rd, buf[:levelsBytelen])
+ if err != nil {
+ return n, err
+ }
+ decodedLen += n
+ if n != levelsBytelen {
+ return n, fmt.Errorf("parquet: expected to read %d
bytes for levels, got %d", levelsBytelen, n)
}
}
- if _, err := p.decompress(p.r, lenCompressed-levelsBytelen,
buf[levelsBytelen:]); err != nil {
- return err
+ values, err := p.decompress(p.r, lenCompressed-levelsBytelen,
buf[levelsBytelen:])
+ if err != nil {
+ return decodedLen, err
}
- return nil
+ decodedLen += len(values)
+ if len(values) != lenUncompressed-decodedLen {
Review Comment:
This is the blocking issue. `decodedLen += len(values)` on the line above
already makes `decodedLen == levelsBytelen + len(values)`, so `lenUncompressed
- decodedLen` here reduces to `lenUncompressed - levelsBytelen - len(values)` —
which is `0` for any valid page. That turns this into `len(values) != 0`, i.e.
it rejects **every** non-empty compressed DataPageV2. (The encrypted sibling
gets it right: `len(decoded) != lenUncompressed-levelsBytelen`.)
Suggested fix:
```go
decodedLen += len(values)
if decodedLen != lenUncompressed {
return decodedLen, fmt.Errorf("parquet: metadata said %d bytes
uncompressed data page, got %d bytes", lenUncompressed, decodedLen)
}
```
I verified on this branch with a valid compressed V2 page (Snappy, no
levels): `Next()` returns false with the self-contradictory `parquet: metadata
said 128 bytes uncompressed data page, got 128 bytes`. The new tests here don't
catch it because they only exercise the mismatched and encrypted paths — please
also add a happy-path test for a valid *compressed* DataPageV2, since the
existing `TestDataPageV2` is uncompressed and `TestCompression` is V1-only.
--
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]
