This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new c03882ff fix(parquet): reject short page writes (#1050)
c03882ff is described below
commit c03882ffcdb819866da2496e70f8f86d11d7c577
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:03:19 2026 +0200
fix(parquet): reject short page writes (#1050)
## Summary
- return `io.ErrShortWrite` when a direct dictionary or data page body
write accepts fewer bytes than requested
- avoid updating page counters and indexes after an incomplete body
write
- cover both direct page paths with a short-writing sink
## Testing
- `go test ./parquet/file`
---
parquet/file/page_writer.go | 12 ++--
parquet/file/page_writer_internal_test.go | 109 ++++++++++++++++++++++++++++++
2 files changed, 115 insertions(+), 6 deletions(-)
diff --git a/parquet/file/page_writer.go b/parquet/file/page_writer.go
index 7ab2b024..144d39b7 100644
--- a/parquet/file/page_writer.go
+++ b/parquet/file/page_writer.go
@@ -320,9 +320,6 @@ func (pw *serializedPageWriter) WriteDictionaryPage(page
*DictionaryPage) (int64
pageHdr.DataPageHeaderV2 = nil
startPos := pw.sink.Tell()
- if pw.dictPageOffset == 0 {
- pw.dictPageOffset = int64(startPos)
- }
if pw.metaEncryptor != nil {
if err := pw.updateEncryption(encryption.DictPageHeaderModule);
err != nil {
@@ -339,6 +336,9 @@ func (pw *serializedPageWriter) WriteDictionaryPage(page
*DictionaryPage) (int64
}
written += headerSize
+ if pw.dictPageOffset == 0 {
+ pw.dictPageOffset = int64(startPos)
+ }
pw.totalUncompressed += int64(uncompressed + headerSize)
pw.totalCompressed = int64(written)
@@ -383,9 +383,6 @@ func (pw *serializedPageWriter) WriteDataPage(page
DataPage) (int64, error) {
}
startPos := pw.sink.Tell()
- if pw.pageOrdinal == 0 {
- pw.dataPageOffset = int64(startPos)
- }
if pw.metaEncryptor != nil {
if err := pw.updateEncryption(encryption.DataPageHeaderModule);
err != nil {
@@ -401,6 +398,9 @@ func (pw *serializedPageWriter) WriteDataPage(page
DataPage) (int64, error) {
return int64(written), err
}
written += headerSize
+ if pw.pageOrdinal == 0 {
+ pw.dataPageOffset = int64(startPos)
+ }
// collect page index
if pw.columnIndexBuilder != nil {
diff --git a/parquet/file/page_writer_internal_test.go
b/parquet/file/page_writer_internal_test.go
new file mode 100644
index 00000000..38c5a236
--- /dev/null
+++ b/parquet/file/page_writer_internal_test.go
@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package file
+
+import (
+ "bytes"
+ "errors"
+ "io"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/compress"
+ "github.com/stretchr/testify/require"
+)
+
+type shortPageSink struct {
+ data []byte
+ pos int64
+ headerShort bool
+}
+
+func (s *shortPageSink) Tell() int64 { return s.pos }
+
+func (s *shortPageSink) Write(p []byte) (int, error) {
+ n := len(p)
+ if s.headerShort || bytes.Equal(p, s.data) {
+ n--
+ }
+ s.pos += int64(n)
+ if n != len(p) {
+ return n, io.ErrShortWrite
+ }
+ return n, nil
+}
+
+func TestSerializedPageWriterRejectsShortHeaderWrites(t *testing.T) {
+ sink := &shortPageSink{headerShort: true, pos: 17}
+ writer, err := NewPageWriter(sink, compress.Codecs.Uncompressed,
+ compress.DefaultCompressionLevel, nil, -1, -1,
memory.DefaultAllocator, false, nil, nil)
+ require.NoError(t, err)
+
+ buf := memory.NewBufferBytes([]byte("page body"))
+ defer buf.Release()
+ _, err = writer.WriteDictionaryPage(NewDictionaryPage(buf, 1,
parquet.Encodings.Plain))
+ require.ErrorIs(t, err, io.ErrShortWrite)
+ serialized := writer.(*serializedPageWriter)
+ require.Zero(t, serialized.NumValues())
+ require.Zero(t, serialized.DictionaryPageOffset())
+}
+
+func TestSerializedPageWriterRejectsShortWrites(t *testing.T) {
+ data := []byte("page body")
+
+ tests := []struct {
+ name string
+ wantWritten int64
+ write func(PageWriter, *memory.Buffer) (int64, error)
+ }{
+ {
+ name: "dictionary page",
+ wantWritten: 0,
+ write: func(writer PageWriter, data *memory.Buffer)
(int64, error) {
+ return
writer.WriteDictionaryPage(NewDictionaryPage(data, 1, parquet.Encodings.Plain))
+ },
+ },
+ {
+ name: "data page",
+ wantWritten: int64(len(data) - 1),
+ write: func(writer PageWriter, data *memory.Buffer)
(int64, error) {
+ return writer.WriteDataPage(NewDataPageV1(data,
1, parquet.Encodings.Plain,
+ parquet.Encodings.RLE,
parquet.Encodings.RLE, int32(data.Len())))
+ },
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ sink := &shortPageSink{data: data, pos: 17}
+ writer, err := NewPageWriter(sink,
compress.Codecs.Uncompressed,
+ compress.DefaultCompressionLevel, nil, -1, -1,
memory.DefaultAllocator, false, nil, nil)
+ require.NoError(t, err)
+
+ buf := memory.NewBufferBytes(data)
+ defer buf.Release()
+ written, err := tt.write(writer, buf)
+ require.True(t, errors.Is(err, io.ErrShortWrite))
+ require.Equal(t, tt.wantWritten, written)
+ serialized := writer.(*serializedPageWriter)
+ require.Zero(t, serialized.NumValues())
+ require.Zero(t, serialized.DictionaryPageOffset())
+ require.Zero(t, serialized.DataPageoffset())
+ })
+ }
+}
