Repository: segmentio/parquet-go
Pull request #430 (2022-12-05 20:44:26)

add parquet.DropDuplicatedRows
Based on https://github.com/segmentio/parquet-go/pull/427, this PR adds the ability to configure the `parquet.SortingWriter` to drop duplicated rows.

As part of this PR, I also revisited the way sorting options are configured on writers and row groups.


Patch:
diff --git a/README.md b/README.md
index 5dda8c2..ecf8743 100644
--- a/README.md
+++ b/README.md
@@ -246,9 +246,11 @@ written to a parquet file:
 type RowType struct { FirstName, LastName string }
 
 buffer := parquet.NewGenericBuffer[RowType](
-    parquet.SortingColumns(
-        parquet.Ascending("LastName"),
-        parquet.Ascending("FistName"),
+    parquet.SortingRowGroupConfig(
+        parquet.SortingColumns(
+            parquet.Ascending("LastName"),
+            parquet.Ascending("FistName"),
+        ),
     ),
 )
 
diff --git a/buffer.go b/buffer.go
index c29a027..cabb935 100644
--- a/buffer.go
+++ b/buffer.go
@@ -57,7 +57,7 @@ func (buf *Buffer) configure(schema *Schema) {
 	if schema == nil {
 		return
 	}
-	sortingColumns := buf.config.SortingColumns
+	sortingColumns := buf.config.Sorting.SortingColumns
 	buf.sorted = make([]ColumnBuffer, len(sortingColumns))
 
 	forEachLeafColumnOf(schema, func(leaf leafColumn) {
@@ -145,7 +145,7 @@ func (buf *Buffer) Schema() *Schema { return buf.schema }
 //
 // The sorting order is configured by passing a SortingColumns option when
 // constructing the buffer.
-func (buf *Buffer) SortingColumns() []SortingColumn { return buf.config.SortingColumns }
+func (buf *Buffer) SortingColumns() []SortingColumn { return buf.config.Sorting.SortingColumns }
 
 // Len returns the number of rows written to the buffer.
 func (buf *Buffer) Len() int {
diff --git a/config.go b/config.go
index 9885544..a6d547e 100644
--- a/config.go
+++ b/config.go
@@ -206,9 +206,9 @@ type WriterConfig struct {
 	MaxRowsPerRowGroup   int64
 	KeyValueMetadata     map[string]string
 	Schema               *Schema
-	SortingColumns       []SortingColumn
 	BloomFilters         []BloomFilterColumn
 	Compression          compress.Codec
+	Sorting              SortingConfig
 }
 
 // DefaultWriterConfig returns a new WriterConfig value initialized with the
@@ -255,6 +255,7 @@ func (c *WriterConfig) ConfigureWriter(config *WriterConfig) {
 			keyValueMetadata[k] = v
 		}
 	}
+
 	*config = WriterConfig{
 		CreatedBy:            coalesceString(c.CreatedBy, config.CreatedBy),
 		ColumnPageBuffers:    coalescePageBufferPool(c.ColumnPageBuffers, config.ColumnPageBuffers),
@@ -266,9 +267,9 @@ func (c *WriterConfig) ConfigureWriter(config *WriterConfig) {
 		MaxRowsPerRowGroup:   config.MaxRowsPerRowGroup,
 		KeyValueMetadata:     keyValueMetadata,
 		Schema:               coalesceSchema(c.Schema, config.Schema),
-		SortingColumns:       coalesceSortingColumns(c.SortingColumns, config.SortingColumns),
 		BloomFilters:         coalesceBloomFilters(c.BloomFilters, config.BloomFilters),
 		Compression:          coalesceCompression(c.Compression, config.Compression),
+		Sorting:              coalesceSortingConfig(c.Sorting, config.Sorting),
 	}
 }
 
@@ -293,8 +294,8 @@ func (c *WriterConfig) Validate() error {
 //	})
 type RowGroupConfig struct {
 	ColumnBufferCapacity int
-	SortingColumns       []SortingColumn
 	Schema               *Schema
+	Sorting              SortingConfig
 }
 
 // DefaultRowGroupConfig returns a new RowGroupConfig value initialized with the
@@ -333,8 +334,60 @@ func (c *RowGroupConfig) Apply(options ...RowGroupOption) {
 func (c *RowGroupConfig) ConfigureRowGroup(config *RowGroupConfig) {
 	*config = RowGroupConfig{
 		ColumnBufferCapacity: coalesceInt(c.ColumnBufferCapacity, config.ColumnBufferCapacity),
-		SortingColumns:       coalesceSortingColumns(c.SortingColumns, config.SortingColumns),
 		Schema:               coalesceSchema(c.Schema, config.Schema),
+		Sorting:              coalesceSortingConfig(c.Sorting, config.Sorting),
+	}
+}
+
+// The SortingConfig type carries configuration options for parquet row groups.
+//
+// SortingConfig implements the SortingOption interface so it can be used
+// directly as argument to the NewSortingWriter function when needed,
+// for example:
+//
+//	buffer := parquet.NewSortingWriter[Row](
+//		parquet.SortingWriterConfig(
+//			parquet.DropDuplicatedRows(true),
+//		),
+//	})
+type SortingConfig struct {
+	SortingColumns     []SortingColumn
+	DropDuplicatedRows bool
+}
+
+// DefaultSortingConfig returns a new SortingConfig value initialized with the
+// default row group configuration.
+func DefaultSortingConfig() *SortingConfig {
+	return &SortingConfig{
+		DropDuplicatedRows: false,
+	}
+}
+
+// NewSortingConfig constructs a new sorting configuration applying the
+// options passed as arguments.
+//
+// The function returns an non-nil error if some of the options carried invalid
+// configuration values.
+func NewSortingConfig(options ...SortingOption) (*SortingConfig, error) {
+	config := DefaultSortingConfig()
+	config.Apply(options...)
+	return config, config.Validate()
+}
+
+func (c *SortingConfig) Validate() error {
+	return nil
+}
+
+func (c *SortingConfig) Apply(options ...SortingOption) {
+	for _, opt := range options {
+		opt.ConfigureSorting(c)
+	}
+}
+
+func (c *SortingConfig) ConfigureSorting(config *SortingConfig) {
+	*config = SortingConfig{
+		SortingColumns:     coalesceSortingColumns(c.SortingColumns, config.SortingColumns),
+		DropDuplicatedRows: c.DropDuplicatedRows,
 	}
 }
 
@@ -362,6 +415,12 @@ type RowGroupOption interface {
 	ConfigureRowGroup(*RowGroupConfig)
 }
 
+// SortingOption is an interface implemented by types that carry configuration
+// options for parquet sorting writers.
+type SortingOption interface {
+	ConfigureSorting(*SortingConfig)
+}
+
 // SkipPageIndex is a file configuration option which prevents automatically
 // reading the page index when opening a parquet file, when set to true. This is
 // useful as an optimization when programs know that they will not need to
@@ -532,6 +591,13 @@ func Compression(codec compress.Codec) WriterOption {
 	return writerOption(func(config *WriterConfig) { config.Compression = codec })
 }
 
+// SortingWriterConfig is a writer option which applies configuration specific
+// to sorting writers.
+func SortingWriterConfig(options ...SortingOption) WriterOption {
+	options = append([]SortingOption{}, options...)
+	return writerOption(func(config *WriterConfig) { config.Sorting.Apply(options...) })
+}
+
 // ColumnBufferCapacity creates a configuration option which defines the size of
 // row group column buffers.
 //
@@ -540,32 +606,37 @@ func ColumnBufferCapacity(size int) RowGroupOption {
 	return rowGroupOption(func(config *RowGroupConfig) { config.ColumnBufferCapacity = size })
 }
 
+// SortingRowGroupConfig is a row group option which applies configuration
+// specific sorting row groups.
+func SortingRowGroupConfig(options ...SortingOption) RowGroupOption {
+	options = append([]SortingOption{}, options...)
+	return rowGroupOption(func(config *RowGroupConfig) { config.Sorting.Apply(options...) })
+}
+
 // SortingColumns creates a configuration option which defines the sorting order
 // of columns in a row group.
 //
 // The order of sorting columns passed as argument defines the ordering
 // hierarchy; when elements are equal in the first column, the second column is
 // used to order rows, etc...
-func SortingColumns(columns ...SortingColumn) interface {
-	RowGroupOption
-	WriterOption
-} {
+func SortingColumns(columns ...SortingColumn) SortingOption {
 	// Make a copy so that we do not retain the input slice generated implicitly
 	// for the variable argument list, and also avoid having a nil slice when
 	// the option is passed with no sorting columns, so we can differentiate it
 	// from it not being passed.
 	columns = append([]SortingColumn{}, columns...)
-	return sortingColumns(columns)
-}
-
-type sortingColumns []SortingColumn
-
-func (columns sortingColumns) ConfigureRowGroup(config *RowGroupConfig) {
-	config.SortingColumns = columns
+	return sortingOption(func(config *SortingConfig) { config.SortingColumns = columns })
 }
 
-func (columns sortingColumns) ConfigureWriter(config *WriterConfig) {
-	config.SortingColumns = columns
+// DropDuplicatedRows configures whether a sorting writer will keep or remove
+// duplicated rows.
+//
+// Two rows are considered duplicates if the values of their all their sorting
+// columns are equal.
+//
+// Defaults to false
+func DropDuplicatedRows(drop bool) SortingOption {
+	return sortingOption(func(config *SortingConfig) { config.DropDuplicatedRows = drop })
 }
 
 type fileOption func(*FileConfig)
@@ -584,6 +655,10 @@ type rowGroupOption func(*RowGroupConfig)
 
 func (opt rowGroupOption) ConfigureRowGroup(config *RowGroupConfig) { opt(config) }
 
+type sortingOption func(*SortingConfig)
+
+func (opt sortingOption) ConfigureSorting(config *SortingConfig) { opt(config) }
+
 func coalesceInt(i1, i2 int) int {
 	if i1 != 0 {
 		return i1
@@ -633,6 +708,13 @@ func coalesceSortingColumns(s1, s2 []SortingColumn) []SortingColumn {
 	return s2
 }
 
+func coalesceSortingConfig(c1, c2 SortingConfig) SortingConfig {
+	return SortingConfig{
+		SortingColumns:     coalesceSortingColumns(c1.SortingColumns, c2.SortingColumns),
+		DropDuplicatedRows: c1.DropDuplicatedRows,
+	}
+}
+
 func coalesceBloomFilters(f1, f2 []BloomFilterColumn) []BloomFilterColumn {
 	if f1 != nil {
 		return f1
@@ -722,4 +804,5 @@ var (
 	_ ReaderOption   = (*ReaderConfig)(nil)
 	_ WriterOption   = (*WriterConfig)(nil)
 	_ RowGroupOption = (*RowGroupConfig)(nil)
+	_ SortingOption  = (*SortingConfig)(nil)
 )
diff --git a/dedupe.go b/dedupe.go
new file mode 100644
index 0000000..9355488
--- /dev/null
+++ b/dedupe.go
@@ -0,0 +1,111 @@
+package parquet
+
+// DedupeRowReader constructs a row reader which drops duplicated consecutive
+// rows, according to the comparator function passed as argument.
+//
+// If the underlying reader produces a sequence of rows sorted by the same
+// comparison predicate, the output is guaranteed to produce unique rows only.
+func DedupeRowReader(reader RowReader, compare func(Row, Row) int) RowReader {
+	return &dedupeRowReader{reader: reader, compare: compare}
+}
+
+type dedupeRowReader struct {
+	reader  RowReader
+	compare func(Row, Row) int
+	dedupe
+}
+
+func (d *dedupeRowReader) ReadRows(rows []Row) (int, error) {
+	for {
+		n, err := d.reader.ReadRows(rows)
+		n = d.deduplicate(rows[:n], d.compare)
+
+		if n > 0 || err != nil {
+			return n, err
+		}
+	}
+}
+
+// DedupeRowWriter constructs a row writer which drops duplicated consecutive
+// rows, according to the comparator function passed as argument.
+//
+// If the writer is given a sequence of rows sorted by the same comparison
+// predicate, the output is guaranteed to contain unique rows only.
+func DedupeRowWriter(writer RowWriter, compare func(Row, Row) int) RowWriter {
+	return &dedupeRowWriter{writer: writer, compare: compare}
+}
+
+type dedupeRowWriter struct {
+	writer  RowWriter
+	compare func(Row, Row) int
+	dedupe
+	rows []Row
+}
+
+func (d *dedupeRowWriter) WriteRows(rows []Row) (int, error) {
+	// We need to make a copy because we cannot modify the rows slice received
+	// as argument to respect the RowWriter contract.
+	d.rows = append(d.rows[:0], rows...)
+	defer func() {
+		for i := range d.rows {
+			d.rows[i] = Row{}
+		}
+	}()
+
+	if n := d.deduplicate(d.rows, d.compare); n > 0 {
+		w, err := d.writer.WriteRows(d.rows[:n])
+		if err != nil {
+			return w, err
+		}
+	}
+
+	// Return the number of rows received instead of the number of deduplicated
+	// rows actually written to the underlying writer because we have to repsect
+	// the RowWriter contract.
+	return len(rows), nil
+}
+
+type dedupe struct {
+	alloc   rowAllocator
+	lastRow Row
+	uniq    []Row
+	dupe    []Row
+}
+
+func (d *dedupe) reset() {
+	d.alloc.reset()
+	d.lastRow = d.lastRow[:0]
+}
+
+func (d *dedupe) deduplicate(rows []Row, compare func(Row, Row) int) int {
+	defer func() {
+		for i := range d.uniq {
+			d.uniq[i] = Row{}
+		}
+		for i := range d.dupe {
+			d.dupe[i] = Row{}
+		}
+		d.uniq = d.uniq[:0]
+		d.dupe = d.dupe[:0]
+	}()
+
+	lastRow := d.lastRow
+
+	for _, row := range rows {
+		if len(lastRow) != 0 && compare(row, lastRow) == 0 {
+			d.dupe = append(d.dupe, row)
+		} else {
+			lastRow = row
+			d.uniq = append(d.uniq, row)
+		}
+	}
+
+	rows = rows[:0]
+	rows = append(rows, d.uniq...)
+	rows = append(rows, d.dupe...)
+
+	d.alloc.reset()
+	d.alloc.capture(lastRow)
+	d.lastRow = append(d.lastRow[:0], lastRow...)
+	return len(d.uniq)
+}
diff --git a/row.go b/row.go
index 3b29856..96651e9 100644
--- a/row.go
+++ b/row.go
@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+
+	"github.com/segmentio/parquet-go/internal/unsafecast"
 )
 
 const (
@@ -782,3 +784,46 @@ func reconstructFuncOfLeaf(columnIndex int16, node Node) (int16, reconstructFunc
 		return row[1:], err
 	}
 }
+
+// rowAllocator is a memory allocator used to make copy of rows referencing
+// memory buffers that parquet-go does not have ownership of.
+//
+// This type is used in the implementation of a various readers and writers that
+// need to capture rows passed to the ReadRows/WriteRows methods.
+//
+// See: RowBuffer, DedupeRowReader, DedupeRowWriter
+type rowAllocator struct{ buffer []byte }
+
+func (a *rowAllocator) reset() {
+	a.buffer = a.buffer[:0]
+}
+
+func (a *rowAllocator) capture(row Row) {
+	for i, v := range row {
+		switch kind := v.Kind(); kind {
+		case ByteArray, FixedLenByteArray:
+			row[i].ptr = unsafecast.AddressOfBytes(a.copyBytes(v.byteArray()))
+		}
+	}
+}
+
+func (a *rowAllocator) copyBytes(v []byte) []byte {
+	if free := cap(a.buffer) - len(a.buffer); free < len(v) {
+		newCap := 2 * cap(a.buffer)
+		if newCap == 0 {
+			newCap = defaultReadBufferSize
+		}
+		for newCap < len(v) {
+			newCap *= 2
+		}
+		a.buffer = make([]byte, 0, newCap)
+	}
+
+	i := len(a.buffer)
+	j := len(a.buffer) + len(v)
+	a.buffer = a.buffer[:j]
+
+	b := a.buffer[i:j:j]
+	copy(b, v)
+	return b
+}
diff --git a/row_buffer.go b/row_buffer.go
index 0d45778..2b36027 100644
--- a/row_buffer.go
+++ b/row_buffer.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/encoding"
-	"github.com/segmentio/parquet-go/internal/unsafecast"
 )
 
 // RowBuffer is an implementation of the RowGroup interface which stores parquet
@@ -23,9 +22,9 @@ import (
 //
 // RowBuffer values are not safe to use concurrently from multiple goroutines.
 type RowBuffer[T any] struct {
+	alloc   rowAllocator
 	schema  *Schema
 	sorting []SortingColumn
-	buffer  []byte
 	rows    []Row
 	numRows int
 	compare func(Row, Row) int
@@ -50,8 +49,8 @@ func NewRowBuffer[T any](options ...RowGroupOption) *RowBuffer[T] {
 
 	return &RowBuffer[T]{
 		schema:  config.Schema,
-		sorting: config.SortingColumns,
-		compare: config.Schema.Comparator(config.SortingColumns...),
+		sorting: config.Sorting.SortingColumns,
+		compare: config.Schema.Comparator(config.Sorting.SortingColumns...),
 	}
 }
 
@@ -62,7 +61,7 @@ func (buf *RowBuffer[T]) Reset() {
 			row[i] = Value{}
 		}
 	}
-	buf.buffer = buf.buffer[:0]
+	buf.alloc.reset()
 	buf.numRows = 0
 }
 
@@ -153,7 +152,7 @@ func (buf *RowBuffer[T]) Write(rows []T) (int, error) {
 	for i := range rows {
 		bufRow := buf.rows[buf.numRows][:0]
 		bufRow = buf.schema.Deconstruct(bufRow, &rows[i])
-		buf.capture(bufRow)
+		buf.alloc.capture(bufRow)
 		buf.rows[buf.numRows] = bufRow
 		buf.numRows++
 	}
@@ -169,7 +168,7 @@ func (buf *RowBuffer[T]) WriteRows(rows []Row) (int, error) {
 	for _, row := range rows {
 		bufRow := buf.rows[buf.numRows][:0]
 		bufRow = append(bufRow, row...)
-		buf.capture(bufRow)
+		buf.alloc.capture(bufRow)
 		buf.rows[buf.numRows] = bufRow
 		buf.numRows++
 	}
@@ -177,36 +176,6 @@ func (buf *RowBuffer[T]) WriteRows(rows []Row) (int, error) {
 	return len(rows), nil
 }
 
-func (buf *RowBuffer[T]) copyBytes(v []byte) []byte {
-	if free := cap(buf.buffer) - len(buf.buffer); free < len(v) {
-		newCap := 2 * cap(buf.buffer)
-		if newCap == 0 {
-			newCap = defaultReadBufferSize
-		}
-		for newCap < len(v) {
-			newCap *= 2
-		}
-		buf.buffer = make([]byte, 0, newCap)
-	}
-
-	i := len(buf.buffer)
-	j := len(buf.buffer) + len(v)
-	buf.buffer = buf.buffer[:j]
-
-	b := buf.buffer[i:j:j]
-	copy(b, v)
-	return b
-}
-
-func (buf *RowBuffer[T]) capture(row Row) {
-	for i, v := range row {
-		switch kind := v.Kind(); kind {
-		case ByteArray, FixedLenByteArray:
-			row[i].ptr = unsafecast.AddressOfBytes(buf.copyBytes(v.byteArray()))
-		}
-	}
-}
-
 func (buf *RowBuffer[T]) reserve(n int) {
 	if newLen := buf.numRows + n; newLen > len(buf.rows) {
 		bufLen := 2 * len(buf.rows)
@@ -498,8 +467,16 @@ func (r *rowBufferRows) ReadRows(rows []Row) (n int, err error) {
 	return n, err
 }
 
+func (r *rowBufferRows) WriteRowsTo(w RowWriter) (int64, error) {
+	n, err := w.WriteRows(r.rows[r.index:])
+	r.index += n
+	return int64(n), err
+}
+
 var (
 	_ RowGroup       = (*RowBuffer[any])(nil)
 	_ RowWriter      = (*RowBuffer[any])(nil)
 	_ sort.Interface = (*RowBuffer[any])(nil)
+
+	_ RowWriterTo = (*rowBufferRows)(nil)
 )
diff --git a/row_group.go b/row_group.go
index b3a9158..b2cb141 100644
--- a/row_group.go
+++ b/row_group.go
@@ -196,7 +196,7 @@ func MergeRowGroups(rowGroups []RowGroup, options ...RowGroupOption) (RowGroup,
 		}
 	}
 
-	m := &mergedRowGroup{sorting: config.SortingColumns}
+	m := &mergedRowGroup{sorting: config.Sorting.SortingColumns}
 	m.init(schema, mergedRowGroups)
 
 	if len(m.sorting) == 0 {
diff --git a/sort.go b/sort.go
index 5517cc2..465d1ca 100644
--- a/sort.go
+++ b/sort.go
@@ -34,7 +34,7 @@ func (c *SortConfig) ConfigureSort(config *SortConfig) {
 //
 // Defaults to zero, which represents a non-repeated column.
 func SortMaxRepetitionLevel(level int) SortOption {
-	return sortingOption(func(c *SortConfig) { c.MaxRepetitionLevel = level })
+	return sortOption(func(c *SortConfig) { c.MaxRepetitionLevel = level })
 }
 
 // SortMaxDefinitionLevel constructs a configuration option which sets the
@@ -42,7 +42,7 @@ func SortMaxRepetitionLevel(level int) SortOption {
 //
 // Defaults to zero, which represents a non-nullable column.
 func SortMaxDefinitionLevel(level int) SortOption {
-	return sortingOption(func(c *SortConfig) { c.MaxDefinitionLevel = level })
+	return sortOption(func(c *SortConfig) { c.MaxDefinitionLevel = level })
 }
 
 // SortDescending constructs a configuration option which inverts the order of a
@@ -50,7 +50,7 @@ func SortMaxDefinitionLevel(level int) SortOption {
 //
 // Defaults to false, which means values are sorted in ascending order.
 func SortDescending(descending bool) SortOption {
-	return sortingOption(func(c *SortConfig) { c.Descending = descending })
+	return sortOption(func(c *SortConfig) { c.Descending = descending })
 }
 
 // SortNullsFirst constructs a configuration option which places the null values
@@ -58,7 +58,7 @@ func SortDescending(descending bool) SortOption {
 //
 // Defaults to false, which means null values are placed last.
 func SortNullsFirst(nullsFirst bool) SortOption {
-	return sortingOption(func(c *SortConfig) { c.NullsFirst = nullsFirst })
+	return sortOption(func(c *SortConfig) { c.NullsFirst = nullsFirst })
 }
 
 // SortOption is an interface implemented by types that carry configuration
@@ -67,9 +67,9 @@ type SortOption interface {
 	ConfigureSort(*SortConfig)
 }
 
-type sortingOption func(*SortConfig)
+type sortOption func(*SortConfig)
 
-func (f sortingOption) ConfigureSort(c *SortConfig) { f(c) }
+func (f sortOption) ConfigureSort(c *SortConfig) { f(c) }
 
 // SortFunc is a function type which compares two sets of column values.
 //
diff --git a/sorting.go b/sorting.go
index 6243d8b..2311a0b 100644
--- a/sorting.go
+++ b/sorting.go
@@ -29,7 +29,8 @@ type SortingWriter[T any] struct {
 	output  *GenericWriter[T]
 	maxRows int64
 	numRows int64
-	sorting []SortingColumn
+	sorting SortingConfig
+	dedupe  dedupe
 }
 
 // NewSortingWriter constructs a new sorting writer which writes a parquet file
@@ -57,8 +58,8 @@ func NewSortingWriter[T any](output io.Writer, sortRowCount int64, options ...Wr
 
 	return &SortingWriter[T]{
 		rows: NewRowBuffer[T](&RowGroupConfig{
-			Schema:         config.Schema,
-			SortingColumns: config.SortingColumns,
+			Schema:  config.Schema,
+			Sorting: config.Sorting,
 		}),
 		buffer: buffer,
 		writer: NewGenericWriter[T](buffer, &WriterConfig{
@@ -69,12 +70,12 @@ func NewSortingWriter[T any](output io.Writer, sortRowCount int64, options ...Wr
 			WriteBufferSize:      config.WriteBufferSize,
 			DataPageVersion:      config.DataPageVersion,
 			Schema:               config.Schema,
-			SortingColumns:       config.SortingColumns,
 			Compression:          config.Compression,
+			Sorting:              config.Sorting,
 		}),
 		output:  NewGenericWriter[T](output, config),
 		maxRows: sortRowCount,
-		sorting: config.SortingColumns,
+		sorting: config.Sorting,
 	}
 }
 
@@ -117,15 +118,23 @@ func (w *SortingWriter[T]) Flush() error {
 
 	m, err := MergeRowGroups(f.RowGroups(),
 		&RowGroupConfig{
-			Schema:         w.Schema(),
-			SortingColumns: w.sorting,
+			Schema:  w.Schema(),
+			Sorting: w.sorting,
 		},
 	)
 	if err != nil {
 		return err
 	}
 
-	if _, err := w.output.WriteRowGroup(m); err != nil {
+	rows := m.Rows()
+	defer rows.Close()
+
+	reader := RowReader(rows)
+	if w.sorting.DropDuplicatedRows {
+		reader = DedupeRowReader(rows, w.rows.compare)
+	}
+
+	if _, err := CopyRows(w.output, reader); err != nil {
 		return err
 	}
 
@@ -191,7 +200,15 @@ func (w *SortingWriter[T]) sortAndWriteBufferedRows() error {
 	defer w.rows.Reset()
 	sort.Sort(w.rows)
 
-	n, err := w.writer.WriteRowGroup(w.rows)
+	if w.sorting.DropDuplicatedRows {
+		w.rows.numRows = w.dedupe.deduplicate(w.rows.rows[:w.rows.numRows], w.rows.compare)
+		defer w.dedupe.reset()
+	}
+
+	rows := w.rows.Rows()
+	defer rows.Close()
+
+	n, err := CopyRows(w.writer, rows)
 	if err != nil {
 		return err
 	}
diff --git a/writer.go b/writer.go
index 0ee2e24..6bb76fd 100644
--- a/writer.go
+++ b/writer.go
@@ -274,7 +274,7 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 		w.metadata = append(w.metadata, format.KeyValue{Key: k, Value: v})
 	}
 	sortKeyValueMetadata(w.metadata)
-	w.sortingColumns = make([]format.SortingColumn, len(config.SortingColumns))
+	w.sortingColumns = make([]format.SortingColumn, len(config.Sorting.SortingColumns))
 
 	config.Schema.forEachNode(func(name string, node Node) {
 		nodeType := node.Type()
@@ -388,11 +388,11 @@ func newWriter(output io.Writer, config *WriterConfig) *writer {
 
 		w.columns = append(w.columns, c)
 
-		if sortingIndex := searchSortingColumn(config.SortingColumns, leaf.path); sortingIndex < len(w.sortingColumns) {
+		if sortingIndex := searchSortingColumn(config.Sorting.SortingColumns, leaf.path); sortingIndex < len(w.sortingColumns) {
 			w.sortingColumns[sortingIndex] = format.SortingColumn{
 				ColumnIdx:  int32(leaf.columnIndex),
-				Descending: config.SortingColumns[sortingIndex].Descending(),
-				NullsFirst: config.SortingColumns[sortingIndex].NullsFirst(),
+				Descending: config.Sorting.SortingColumns[sortingIndex].Descending(),
+				NullsFirst: config.Sorting.SortingColumns[sortingIndex].NullsFirst(),
 			}
 		}
 	})


Test patch:
diff --git a/buffer_go18_test.go b/buffer_go18_test.go
index 18acf64..dc2be58 100644
--- a/buffer_go18_test.go
+++ b/buffer_go18_test.go
@@ -238,8 +238,10 @@ func BenchmarkSortGenericBuffer(b *testing.B) {
 	}
 
 	buf := parquet.NewGenericBuffer[Row](
-		parquet.SortingColumns(
-			parquet.Ascending("ID"),
+		parquet.SortingRowGroupConfig(
+			parquet.SortingColumns(
+				parquet.Ascending("ID"),
+			),
 		),
 	)
 
diff --git a/buffer_test.go b/buffer_test.go
index 03c100e..9015f78 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -209,7 +209,11 @@ func TestBuffer(t *testing.T) {
 										parquet.ColumnBufferCapacity(100),
 									}
 									if ordering.sorting != nil {
-										options = append(options, parquet.SortingColumns(ordering.sorting))
+										options = append(options,
+											parquet.SortingRowGroupConfig(
+												parquet.SortingColumns(ordering.sorting),
+											),
+										)
 									}
 
 									content := new(bytes.Buffer)
diff --git a/dedupe_test.go b/dedupe_test.go
new file mode 100644
index 0000000..fd6e1d2
--- /dev/null
+++ b/dedupe_test.go
@@ -0,0 +1,107 @@
+//go:build go1.18
+
+package parquet_test
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/segmentio/parquet-go"
+)
+
+func TestDedupeRowReader(t *testing.T) {
+	type Row struct {
+		Value int32 `parquet:"value"`
+	}
+
+	rows := make([]Row, 1000)
+	for i := range rows {
+		rows[i].Value = int32(i / 3)
+	}
+
+	dedupeMap := make(map[Row]struct{}, len(rows))
+	for _, row := range rows {
+		dedupeMap[row] = struct{}{}
+	}
+
+	dedupeRows := make([]Row, 0, len(dedupeMap))
+	for row := range dedupeMap {
+		dedupeRows = append(dedupeRows, row)
+	}
+
+	sort.Slice(dedupeRows, func(i, j int) bool {
+		return dedupeRows[i].Value < dedupeRows[j].Value
+	})
+
+	buffer1 := parquet.NewRowBuffer[Row]()
+	buffer1.Write(rows)
+
+	buffer1Rows := buffer1.Rows()
+	defer buffer1Rows.Close()
+
+	buffer2 := parquet.NewRowBuffer[Row]()
+
+	_, err := parquet.CopyRows(buffer2,
+		parquet.DedupeRowReader(buffer1Rows,
+			buffer1.Schema().Comparator(parquet.Ascending("value")),
+		),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	reader := parquet.NewGenericRowGroupReader[Row](buffer2)
+	defer reader.Close()
+
+	n, _ := reader.Read(rows)
+	assertRowsEqual(t, dedupeRows, rows[:n])
+}
+
+func TestDedupeRowWriter(t *testing.T) {
+	type Row struct {
+		Value int32 `parquet:"value"`
+	}
+
+	rows := make([]Row, 1000)
+	for i := range rows {
+		rows[i].Value = int32(i / 3)
+	}
+
+	dedupeMap := make(map[Row]struct{}, len(rows))
+	for _, row := range rows {
+		dedupeMap[row] = struct{}{}
+	}
+
+	dedupeRows := make([]Row, 0, len(dedupeMap))
+	for row := range dedupeMap {
+		dedupeRows = append(dedupeRows, row)
+	}
+
+	sort.Slice(dedupeRows, func(i, j int) bool {
+		return dedupeRows[i].Value < dedupeRows[j].Value
+	})
+
+	buffer1 := parquet.NewRowBuffer[Row]()
+	buffer1.Write(rows)
+
+	buffer1Rows := buffer1.Rows()
+	defer buffer1Rows.Close()
+
+	buffer2 := parquet.NewRowBuffer[Row]()
+
+	_, err := parquet.CopyRows(
+		parquet.DedupeRowWriter(buffer2,
+			buffer1.Schema().Comparator(parquet.Ascending("value")),
+		),
+		buffer1Rows,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	reader := parquet.NewGenericRowGroupReader[Row](buffer2)
+	defer reader.Close()
+
+	n, _ := reader.Read(rows)
+	assertRowsEqual(t, dedupeRows, rows[:n])
+}
diff --git a/merge_test.go b/merge_test.go
index 5be47a6..4a832d7 100644
--- a/merge_test.go
+++ b/merge_test.go
@@ -44,8 +44,10 @@ func TestMergeRowGroupsCursorsAreClosed(t *testing.T) {
 
 	schema := parquet.SchemaOf(model{})
 	options := []parquet.RowGroupOption{
-		parquet.SortingColumns(
-			parquet.Ascending(schema.Columns()[0]...),
+		parquet.SortingRowGroupConfig(
+			parquet.SortingColumns(
+				parquet.Ascending(schema.Columns()[0]...),
+			),
 		),
 	}
 
@@ -92,8 +94,10 @@ func BenchmarkMergeRowGroups(b *testing.B) {
 			schema := parquet.SchemaOf(test.model)
 
 			options := []parquet.RowGroupOption{
-				parquet.SortingColumns(
-					parquet.Ascending(schema.Columns()[0]...),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending(schema.Columns()[0]...),
+					),
 				),
 			}
 
@@ -141,8 +145,10 @@ func BenchmarkMergeFiles(b *testing.B) {
 
 			buffer := parquet.NewBuffer(
 				schema,
-				parquet.SortingColumns(
-					parquet.Ascending(schema.Columns()[0]...),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending(schema.Columns()[0]...),
+					),
 				),
 			)
 
diff --git a/row_buffer_test.go b/row_buffer_test.go
index 82c3427..b8184e6 100644
--- a/row_buffer_test.go
+++ b/row_buffer_test.go
@@ -75,7 +75,11 @@ func TestRowBuffer(t *testing.T) {
 							}
 
 							if ordering.sorting != nil {
-								options = append(options, parquet.SortingColumns(ordering.sorting))
+								options = append(options,
+									parquet.SortingRowGroupConfig(
+										parquet.SortingColumns(ordering.sorting),
+									),
+								)
 							}
 
 							content := new(bytes.Buffer)
@@ -270,8 +274,10 @@ func BenchmarkSortRowBuffer(b *testing.B) {
 	}
 
 	buf := parquet.NewRowBuffer[Row](
-		parquet.SortingColumns(
-			parquet.Ascending("ID"),
+		parquet.SortingRowGroupConfig(
+			parquet.SortingColumns(
+				parquet.Ascending("ID"),
+			),
 		),
 	)
 
diff --git a/row_group_test.go b/row_group_test.go
index cb85a97..1c3756c 100644
--- a/row_group_test.go
+++ b/row_group_test.go
@@ -187,15 +187,19 @@ func TestMergeRowGroups(t *testing.T) {
 		{
 			scenario: "row groups sorted by ascending last name",
 			options: []parquet.RowGroupOption{
-				parquet.SortingColumns(
-					parquet.Ascending("LastName"),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending("LastName"),
+					),
 				),
 			},
 			input: []parquet.RowGroup{
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Han", LastName: "Solo"},
@@ -203,8 +207,10 @@ func TestMergeRowGroups(t *testing.T) {
 				),
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Obiwan", LastName: "Kenobi"},
@@ -220,15 +226,19 @@ func TestMergeRowGroups(t *testing.T) {
 		{
 			scenario: "row groups sorted by descending last name",
 			options: []parquet.RowGroupOption{
-				parquet.SortingColumns(
-					parquet.Descending("LastName"),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Descending("LastName"),
+					),
 				),
 			},
 			input: []parquet.RowGroup{
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Descending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Descending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Han", LastName: "Solo"},
@@ -236,8 +246,10 @@ func TestMergeRowGroups(t *testing.T) {
 				),
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Descending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Descending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Obiwan", LastName: "Kenobi"},
@@ -253,17 +265,21 @@ func TestMergeRowGroups(t *testing.T) {
 		{
 			scenario: "row groups sorted by ascending last and first name",
 			options: []parquet.RowGroupOption{
-				parquet.SortingColumns(
-					parquet.Ascending("LastName"),
-					parquet.Ascending("FirstName"),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending("LastName"),
+						parquet.Ascending("FirstName"),
+					),
 				),
 			},
 			input: []parquet.RowGroup{
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
-							parquet.Ascending("FirstName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+								parquet.Ascending("FirstName"),
+							),
 						),
 					},
 					Person{FirstName: "Luke", LastName: "Skywalker"},
@@ -271,9 +287,11 @@ func TestMergeRowGroups(t *testing.T) {
 				),
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
-							parquet.Ascending("FirstName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+								parquet.Ascending("FirstName"),
+							),
 						),
 					},
 					Person{FirstName: "Obiwan", LastName: "Kenobi"},
@@ -292,15 +310,19 @@ func TestMergeRowGroups(t *testing.T) {
 			scenario: "row groups with conversion to a different schema",
 			options: []parquet.RowGroupOption{
 				parquet.SchemaOf(LastNameOnly{}),
-				parquet.SortingColumns(
-					parquet.Ascending("LastName"),
+				parquet.SortingRowGroupConfig(
+					parquet.SortingColumns(
+						parquet.Ascending("LastName"),
+					),
 				),
 			},
 			input: []parquet.RowGroup{
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Han", LastName: "Solo"},
@@ -308,8 +330,10 @@ func TestMergeRowGroups(t *testing.T) {
 				),
 				sortedRowGroup(
 					[]parquet.RowGroupOption{
-						parquet.SortingColumns(
-							parquet.Ascending("LastName"),
+						parquet.SortingRowGroupConfig(
+							parquet.SortingColumns(
+								parquet.Ascending("LastName"),
+							),
 						),
 					},
 					Person{FirstName: "Obiwan", LastName: "Kenobi"},
@@ -318,8 +342,10 @@ func TestMergeRowGroups(t *testing.T) {
 			},
 			output: sortedRowGroup(
 				[]parquet.RowGroupOption{
-					parquet.SortingColumns(
-						parquet.Ascending("LastName"),
+					parquet.SortingRowGroupConfig(
+						parquet.SortingColumns(
+							parquet.Ascending("LastName"),
+						),
 					),
 				},
 				LastNameOnly{LastName: "Solo"},
diff --git a/sorting_test.go b/sorting_test.go
index 20a46c1..3c5236e 100644
--- a/sorting_test.go
+++ b/sorting_test.go
@@ -28,8 +28,10 @@ func TestSortingWriter(t *testing.T) {
 
 	buffer := bytes.NewBuffer(nil)
 	writer := parquet.NewSortingWriter[Row](buffer, 99,
-		parquet.SortingColumns(
-			parquet.Ascending("value"),
+		parquet.SortingWriterConfig(
+			parquet.SortingColumns(
+				parquet.Ascending("value"),
+			),
 		),
 	)
 
@@ -53,3 +55,54 @@ func TestSortingWriter(t *testing.T) {
 
 	assertRowsEqual(t, rows, read)
 }
+
+func TestSortingWriterDropDuplicatedRows(t *testing.T) {
+	type Row struct {
+		Value int32 `parquet:"value"`
+	}
+
+	rows := make([]Row, 1000)
+	for i := range rows {
+		rows[i].Value = int32(i / 2)
+	}
+
+	prng := rand.New(rand.NewSource(0))
+	prng.Shuffle(len(rows), func(i, j int) {
+		rows[i], rows[j] = rows[j], rows[i]
+	})
+
+	buffer := bytes.NewBuffer(nil)
+	writer := parquet.NewSortingWriter[Row](buffer, 99,
+		parquet.SortingWriterConfig(
+			parquet.SortingColumns(
+				parquet.Ascending("value"),
+			),
+			parquet.DropDuplicatedRows(true),
+		),
+	)
+
+	_, err := writer.Write(rows)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	read, err := parquet.Read[Row](bytes.NewReader(buffer.Bytes()), int64(buffer.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sort.Slice(rows, func(i, j int) bool {
+		return rows[i].Value < rows[j].Value
+	})
+
+	n := len(rows) / 2
+	for i := range rows[:n] {
+		rows[i] = rows[2*i]
+	}
+
+	assertRowsEqual(t, rows[:n], read)
+}