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 6de63de9 fix(parquet): row-align V1 pages with offset indexes (#1075)
6de63de9 is described below
commit 6de63de933edf80c96fefc311a461e8b96f9c914
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:16:29 2026 +0200
fix(parquet): row-align V1 pages with offset indexes (#1075)
Parquet pages referenced by an OffsetIndex must begin at row boundaries. The
writer enforced this for repeated DataPageV2 columns, but repeated
DataPageV1
columns still used fixed-size batching when page indexes were enabled, so a
logical row could span two indexed V1 pages.
Route repeated V1 columns through the existing repetition-level-aware
batching
when page indexing is enabled, and apply the same condition to the generated
variable-length writer paths. V1 writes without page indexes retain the
existing fixed-size batching behavior.
Adds a regression test using indexed V1 pages with repeated three-value rows
and a batch size that does not divide the row width, verifying intermediate
and final pages all begin at row boundaries.
Closes #887
---
parquet/file/column_writer.go | 15 +++------
parquet/file/column_writer_test.go | 41 ++++++++++++++++++++++++
parquet/file/column_writer_types.gen.go | 48 +++++++++++++++-------------
parquet/file/column_writer_types.gen.go.tmpl | 24 +++++++-------
4 files changed, 85 insertions(+), 43 deletions(-)
diff --git a/parquet/file/column_writer.go b/parquet/file/column_writer.go
index bc8cd701..b19d1ce9 100644
--- a/parquet/file/column_writer.go
+++ b/parquet/file/column_writer.go
@@ -656,14 +656,9 @@ func (w *columnWriter) Close() (err error) {
func (w *columnWriter) doBatches(total int64, repLevels []int16, action
func(offset, batch int64)) {
batchSize := w.props.WriteBatchSize()
- // if we're writing V1 data pages, have no replevels or the max
replevel is 0 then just
- // use the regular doBatches function.
- //
- // The spec also requires row-aligned pages for V1 when an OffsetIndex
is
- // present (PageIndexEnabled). That gap is on the WriteBatch path and is
- // out of scope for the DataPageV2 row alignment here; it is tracked in
- // https://github.com/apache/arrow-go/issues/887.
- if w.props.DataPageVersion() == parquet.DataPageV1 || repLevels == nil
|| w.descr.MaxRepetitionLevel() == 0 {
+ requiresRowAlignment := w.props.DataPageVersion() != parquet.DataPageV1
||
+ w.props.PageIndexEnabledFor(w.descr.Path())
+ if !requiresRowAlignment || repLevels == nil ||
w.descr.MaxRepetitionLevel() == 0 {
doBatches(total, batchSize, action)
return
}
@@ -692,7 +687,7 @@ func (w *columnWriter) doBatches(total int64, repLevels
[]int16, action func(off
repLevels = repLevels[:total]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages must start
at a row boundary")
+ panic("columnwriter: row-aligned batch writing must start at a
row boundary")
}
// loop by batchSize, but make sure we're ending/starting each batch on
a row boundary
@@ -723,7 +718,7 @@ func doBatches(total, batchSize int64, action func(offset,
batch int64)) {
// alignBatchToRowBoundary adjusts batch so that repLevels[offset+batch] lands
on
// a row boundary (repetition level 0) or the end of the level slice. A
repeated
-// row must never span a DataPageV2 page boundary, so it first shrinks toward
the
+// row must never span a row-aligned page boundary, so it first shrinks toward
the
// previous boundary. If there is no boundary at or before the requested split
-
// the current row is wider than batch - it grows forward to the next one so
the
// whole row stays in a single batch and the caller keeps making progress
rather
diff --git a/parquet/file/column_writer_test.go
b/parquet/file/column_writer_test.go
index a649a8c1..7f66f79f 100644
--- a/parquet/file/column_writer_test.go
+++ b/parquet/file/column_writer_test.go
@@ -222,6 +222,47 @@ func TestDataPageV2RowBoundaries(t *testing.T) {
wr.WriteBatch(values, defLevels, repLevels)
}
+func TestDataPageV1OffsetIndexRowBoundaries(t *testing.T) {
+ sc := schema.NewSchema(schema.MustGroup(schema.NewGroupNode("schema",
parquet.Repetitions.Required, schema.FieldList{
+ schema.Must(schema.ListOf(
+ schema.Must(schema.NewPrimitiveNode("column",
parquet.Repetitions.Optional, parquet.Types.Int32, -1, -1)),
+ parquet.Repetitions.Optional, -1)),
+ }, -1)))
+ descr := sc.Column(0)
+ props := parquet.NewWriterProperties(
+ parquet.WithBatchSize(128),
+ parquet.WithDataPageSize(1024),
+ parquet.WithDataPageVersion(parquet.DataPageV1),
+ parquet.WithPageIndexEnabled(true),
+ parquet.WithDictionaryDefault(false))
+
+ metadata := metadata.NewColumnChunkMetaDataBuilder(props, descr)
+ pager := new(mockpagewriter)
+ defer pager.AssertExpectations(t)
+ pager.On("HasCompressor").Return(false)
+ wr := file.NewColumnChunkWriter(metadata, pager,
props).(*file.Int32ColumnChunkWriter)
+
+ pager.On("WriteDataPage", mock.MatchedBy(func(page file.DataPage) bool {
+ pagev1, ok := page.(*file.DataPageV1)
+ return ok && pagev1.NumValues()%3 == 0
+ })).Return(10, nil)
+ pager.On("Close", false, false).Return(nil).Once()
+
+ values := make([]int32, 1023)
+ defLevels := make([]int16, 1023)
+ repLevels := make([]int16, 1023)
+ for i := range values {
+ values[i] = int32(i)
+ defLevels[i] = 3
+ if i%3 != 0 {
+ repLevels[i] = 1
+ }
+ }
+
+ wr.WriteBatch(values, defLevels, repLevels)
+ assert.NoError(t, wr.Close())
+}
+
type PrimitiveWriterTestSuite struct {
testutils.PrimitiveTypedTest
suite.Suite
diff --git a/parquet/file/column_writer_types.gen.go
b/parquet/file/column_writer_types.gen.go
index 4ae88bf6..d0dc0a5c 100644
--- a/parquet/file/column_writer_types.gen.go
+++ b/parquet/file/column_writer_types.gen.go
@@ -1806,15 +1806,16 @@ func (w *ByteArrayColumnChunkWriter) WriteBatch(values
[]parquet.ByteArray, defL
batchSize := w.props.WriteBatchSize()
maxDefLevel := w.descr.MaxDefinitionLevel()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
- // Repeated DataPageV2 writes align batches on row boundaries using
repLevels
+ // Repeated writes that produce an offset index align batches on row
boundaries
// below. n comes from defLevels/values, not repLevels, so clamp
repLevels to n
// to stop an oversized slice from growing a batch past n (spilling
extra levels
// or reading out of range). Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for
batch to write")
}
@@ -1823,7 +1824,7 @@ func (w *ByteArrayColumnChunkWriter) WriteBatch(values
[]parquet.ByteArray, defL
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages
must start at a row boundary")
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
}
}
@@ -1850,10 +1851,10 @@ func (w *ByteArrayColumnChunkWriter) WriteBatch(values
[]parquet.ByteArray, defL
}
}
- // V2 row-boundary alignment: a repeated row must not span a
page boundary,
+ // Row-boundary alignment: a repeated row must not span an
indexed page boundary,
// so snap the batch onto the nearest row boundary (or keep a
single wide row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
}
if batch < 1 {
@@ -1920,13 +1921,14 @@ func (w *ByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []parquet.
const maxSafeBatchDataSize int64 = 1 << 30 // 1GB
batchSize := w.props.WriteBatchSize()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
n := int64(length)
// Clamp repLevels to n; see WriteBatch. Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for
batch to write")
}
@@ -1935,7 +1937,7 @@ func (w *ByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []parquet.
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages
must start at a row boundary")
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
}
}
@@ -1955,10 +1957,10 @@ func (w *ByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []parquet.
}
}
- // V2 row-boundary alignment: a repeated row must not span a
page boundary,
+ // Row-boundary alignment: a repeated row must not span an
indexed page boundary,
// so snap the batch onto the nearest row boundary (or keep a
single wide row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
}
if batch < 1 {
@@ -2176,15 +2178,16 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatch(values []parquet.FixedLe
batchSize := w.props.WriteBatchSize()
maxDefLevel := w.descr.MaxDefinitionLevel()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
- // Repeated DataPageV2 writes align batches on row boundaries using
repLevels
+ // Repeated writes that produce an offset index align batches on row
boundaries
// below. n comes from defLevels/values, not repLevels, so clamp
repLevels to n
// to stop an oversized slice from growing a batch past n (spilling
extra levels
// or reading out of range). Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for
batch to write")
}
@@ -2193,7 +2196,7 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatch(values []parquet.FixedLe
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages
must start at a row boundary")
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
}
}
@@ -2220,10 +2223,10 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatch(values []parquet.FixedLe
}
}
- // V2 row-boundary alignment: a repeated row must not span a
page boundary,
+ // Row-boundary alignment: a repeated row must not span an
indexed page boundary,
// so snap the batch onto the nearest row boundary (or keep a
single wide row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
}
if batch < 1 {
@@ -2290,13 +2293,14 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []
const maxSafeBatchDataSize int64 = 1 << 30 // 1GB
batchSize := w.props.WriteBatchSize()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() !=
parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
n := int64(length)
// Clamp repLevels to n; see WriteBatch. Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for
batch to write")
}
@@ -2305,7 +2309,7 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages
must start at a row boundary")
+ panic("columnwriter: row-aligned batch writing must
start at a row boundary")
}
}
@@ -2325,10 +2329,10 @@ func (w *FixedLenByteArrayColumnChunkWriter)
WriteBatchSpacedWithError(values []
}
}
- // V2 row-boundary alignment: a repeated row must not span a
page boundary,
+ // Row-boundary alignment: a repeated row must not span an
indexed page boundary,
// so snap the batch onto the nearest row boundary (or keep a
single wide row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset,
batch)
}
if batch < 1 {
diff --git a/parquet/file/column_writer_types.gen.go.tmpl
b/parquet/file/column_writer_types.gen.go.tmpl
index d23bdfa3..30c1e9e3 100644
--- a/parquet/file/column_writer_types.gen.go.tmpl
+++ b/parquet/file/column_writer_types.gen.go.tmpl
@@ -94,15 +94,16 @@ func (w *{{.Name}}ColumnChunkWriter) WriteBatch(values
[]{{.name}}, defLevels, r
batchSize := w.props.WriteBatchSize()
maxDefLevel := w.descr.MaxDefinitionLevel()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() != parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
- // Repeated DataPageV2 writes align batches on row boundaries using repLevels
+ // Repeated writes that produce an offset index align batches on row
boundaries
// below. n comes from defLevels/values, not repLevels, so clamp repLevels
to n
// to stop an oversized slice from growing a batch past n (spilling extra
levels
// or reading out of range). Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for batch to write")
}
@@ -111,7 +112,7 @@ func (w *{{.Name}}ColumnChunkWriter) WriteBatch(values
[]{{.name}}, defLevels, r
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages must start at a row
boundary")
+ panic("columnwriter: row-aligned batch writing must start at a row
boundary")
}
}
@@ -142,10 +143,10 @@ func (w *{{.Name}}ColumnChunkWriter) WriteBatch(values
[]{{.name}}, defLevels, r
}
}
- // V2 row-boundary alignment: a repeated row must not span a page boundary,
+ // Row-boundary alignment: a repeated row must not span an indexed page
boundary,
// so snap the batch onto the nearest row boundary (or keep a single wide
row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset, batch)
}
if batch < 1 {
@@ -231,13 +232,14 @@ func (w *{{.Name}}ColumnChunkWriter)
WriteBatchSpacedWithError(values []{{.name}
const maxSafeBatchDataSize int64 = 1 << 30 // 1GB
batchSize := w.props.WriteBatchSize()
- isV2WithRep := w.props.DataPageVersion() != parquet.DataPageV1 &&
+ requiresRowAlignment := (w.props.DataPageVersion() != parquet.DataPageV1 ||
+ w.props.PageIndexEnabledFor(w.descr.Path())) &&
repLevels != nil && w.descr.MaxRepetitionLevel() > 0
levelOffset := int64(0)
n := int64(length)
// Clamp repLevels to n; see WriteBatch. Mirrors columnWriter.doBatches.
- if isV2WithRep {
+ if requiresRowAlignment {
if int64(len(repLevels)) < n {
panic("columnwriter: not enough repetition levels for batch to write")
}
@@ -246,7 +248,7 @@ func (w *{{.Name}}ColumnChunkWriter)
WriteBatchSpacedWithError(values []{{.name}
}
repLevels = repLevels[:n]
if repLevels[0] != 0 {
- panic("columnwriter: batch writing for V2 data pages must start at a row
boundary")
+ panic("columnwriter: row-aligned batch writing must start at a row
boundary")
}
}
@@ -270,10 +272,10 @@ func (w *{{.Name}}ColumnChunkWriter)
WriteBatchSpacedWithError(values []{{.name}
}
}
- // V2 row-boundary alignment: a repeated row must not span a page boundary,
+ // Row-boundary alignment: a repeated row must not span an indexed page
boundary,
// so snap the batch onto the nearest row boundary (or keep a single wide
row
// whole). See alignBatchToRowBoundary in column_writer.go.
- if isV2WithRep {
+ if requiresRowAlignment {
batch = alignBatchToRowBoundary(repLevels, levelOffset, batch)
}
if batch < 1 {
