Repository: polarsignals/frostdb
Pull request #535 (2023-09-19 15:21:23)

compaction: treat non-overlapping l0 parts as a single row group
This reduces the number of open cursors during compaction and reduces memory usage. On a production WAL (Polar Signals project), the maximum memory usage went from 27GB to 8.5GB (70% reduction). Replay time has also been slightly improved:
```
goos: darwin
goarch: arm64
pkg: github.com/polarsignals/frostdb
          │ benchmain  │           benchpspart            │
          │   sec/op   │   sec/op    vs base              │
Replay-12   67.26 ± 3%   63.82 ± 1%  -5.12% (p=0.002 n=6)

          │  benchmain   │            benchpspart             │
          │     B/op     │     B/op      vs base              │
Replay-12   243.6Gi ± 2%   235.2Gi ± 2%  -3.42% (p=0.004 n=6)

          │  benchmain  │         benchpspart          │
          │  allocs/op  │  allocs/op   vs base         │
Replay-12   2.492G ± 2%   2.500G ± 2%  ~ (p=0.589 n=6)
```


Patch:
diff --git a/compaction.go b/compaction.go
index 5cac9e4..47cd579 100644
--- a/compaction.go
+++ b/compaction.go
@@ -530,7 +530,7 @@ func compactLevel0IntoLevel1(
 	// cause them to be split to different granules, rendering the index
 	// useless.
 	nonOverlapping := make([]*parts.Part, 0, len(level1Parts))
-	partsToCompact := level0Parts
+	partsToCompact := make([]*parts.Part, 0, len(level1Parts))
 	// size and numRows are used to estimate the number of bytes per row so that
 	// we can provide a maximum number of rows to write in writeRowGroups which
 	// will more or less keep the compacted part under the maximum granule size.
@@ -542,6 +542,8 @@ func compactLevel0IntoLevel1(
 		size = stats.level0SizeBefore
 		numRows = stats.level0NumRowsBefore
 	}
+	// Determine which level1Parts need to be compacted based on their overlap
+	// with level0Parts.
 	for _, p1 := range level1Parts {
 		overlapped := false
 		for _, p0 := range level0Parts {
@@ -559,7 +561,38 @@ func compactLevel0IntoLevel1(
 		}
 	}
 
+	// To reduce the number of open cursors at the same time (which helps in
+	// memory usage reduction), find which level0Parts do not overlap with any
+	// other part. These parts can be sorted and read one by one.
+	nonOverlappingL0, overlappingL0, err := parts.FindMaximumNonOverlappingSet(t.table.schema, level0Parts)
+	if err != nil {
+		return nil, err
+	}
 	bufs := make([]dynparquet.DynamicRowGroup, 0, len(level0Parts))
+	if len(nonOverlappingL0) == 1 {
+		// Not worth doing anything if only one part does not overlap.
+		overlappingL0 = append(overlappingL0, nonOverlappingL0[0])
+	} else if len(nonOverlappingL0) > 0 {
+		// nonOverlappingL0 is already sorted.
+		rowGroups := make([]dynparquet.DynamicRowGroup, 0, len(nonOverlappingL0))
+		for _, p := range nonOverlappingL0 {
+			buf, err := p.AsSerializedBuffer(t.table.schema)
+			if err != nil {
+				return nil, err
+			}
+			rowGroups = append(rowGroups, buf.MultiDynamicRowGroup())
+		}
+		// WithAlreadySorted ensures that a parquet.MultiRowGroup is created
+		// here, which is much cheaper than actually merging all these row
+		// groups.
+		merged, err := t.table.schema.MergeDynamicRowGroups(rowGroups, dynparquet.WithAlreadySorted())
+		if err != nil {
+			return nil, err
+		}
+		bufs = append(bufs, merged)
+	}
+
+	partsToCompact = append(partsToCompact, overlappingL0...)
 	for _, p := range partsToCompact {
 		buf, err := p.AsSerializedBuffer(t.table.schema)
 		if err != nil {
@@ -572,7 +605,6 @@ func compactLevel0IntoLevel1(
 		bufs = append(bufs, buf.MultiDynamicRowGroup())
 	}
 
-	cursor := 0
 	merged, err := t.table.schema.MergeDynamicRowGroups(bufs)
 	if err != nil {
 		return nil, err
@@ -609,7 +641,6 @@ func compactLevel0IntoLevel1(
 			if n == 0 {
 				break
 			}
-			cursor += n
 			serBuf, err := dynparquet.ReaderFromBytes(mergedBytes.Bytes())
 			if err != nil {
 				return err
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 2641c30..53e8ffa 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -5,17 +5,21 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"reflect"
+	"regexp"
 	"sort"
 	"strings"
 	"sync"
 	"text/tabwriter"
 
+	"github.com/google/uuid"
 	"github.com/parquet-go/parquet-go"
 	"github.com/parquet-go/parquet-go/compress"
 	"github.com/parquet-go/parquet-go/encoding"
 	"github.com/parquet-go/parquet-go/format"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/polarsignals/frostdb/bufutils"
 	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
 	schemav2pb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha2"
 )
@@ -951,6 +955,175 @@ func (b *Buffer) DynamicRows() DynamicRowReader {
 	return newDynamicRowGroupReader(b, b.fields)
 }
 
+var (
+	matchFirstCap = regexp.MustCompile("(.)([A-Z][a-z]+)")
+	matchAllCap   = regexp.MustCompile("([a-z0-9])([A-Z])")
+)
+
+func ToSnakeCase(str string) string {
+	snake := matchFirstCap.ReplaceAllString(str, "${1}_${2}")
+	snake = matchAllCap.ReplaceAllString(snake, "${1}_${2}")
+	return strings.ToLower(snake)
+}
+
+func ValuesToBuffer(schema *Schema, vals ...any) (*Buffer, error) {
+	dynamicColumns := map[string][]string{}
+	rows := make([]parquet.Row, 0, len(vals))
+
+	findColumn := func(val reflect.Value, col string, v any) any {
+		for i := 0; i < val.NumField(); i++ {
+			if ToSnakeCase(val.Type().Field(i).Name) == col {
+				return val.Field(i).Interface()
+			}
+		}
+		return nil
+	}
+
+	// Collect dynamic columns
+	for _, v := range vals {
+		val := reflect.ValueOf(v)
+		for _, col := range schema.Columns() {
+			cv := findColumn(val, col.Name, v)
+			if cv == nil {
+				continue
+			}
+			switch col.Dynamic {
+			case true:
+				switch reflect.TypeOf(cv).Kind() {
+				case reflect.Struct:
+					dynVals := reflect.ValueOf(cv)
+					for j := 0; j < dynVals.NumField(); j++ {
+						dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(dynVals.Type().Field(j).Name))
+					}
+				case reflect.Slice:
+					dynVals := reflect.ValueOf(cv)
+					for j := 0; j < dynVals.Len(); j++ {
+						pair := reflect.ValueOf(dynVals.Index(j).Interface())
+						dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(fmt.Sprintf("%v", pair.Field(0))))
+					}
+				case reflect.Map:
+					dynVals := reflect.ValueOf(cv)
+					for _, key := range dynVals.MapKeys() {
+						switch key.Kind() {
+						case reflect.String:
+							dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(fmt.Sprintf("%v", key.String())))
+						default:
+							return nil, fmt.Errorf("unsupported dynamic map key type for column for column '%s'", col.Name)
+						}
+					}
+				default:
+					return nil, fmt.Errorf("unsupported dynamic type for column '%s'", col.Name)
+				}
+			}
+		}
+	}
+
+	dynamicColumns = bufutils.Dedupe(dynamicColumns)
+
+	// Create all rows
+	for _, v := range vals {
+		row := []parquet.Value{}
+		val := reflect.ValueOf(v)
+
+		colIdx := 0
+		for _, col := range schema.Columns() {
+			cv := findColumn(val, col.Name, v)
+			if cv == nil {
+				continue
+			}
+			switch col.Dynamic {
+			case true:
+				switch reflect.TypeOf(cv).Kind() {
+				case reflect.Struct:
+					dynVals := reflect.ValueOf(cv)
+					for _, dyncol := range dynamicColumns[col.Name] {
+						found := false
+						for j := 0; j < dynVals.NumField(); j++ {
+							if ToSnakeCase(dynVals.Type().Field(j).Name) == dyncol {
+								row = append(row, parquet.ValueOf(dynVals.Field(j).Interface()).Level(0, 1, colIdx))
+								colIdx++
+								found = true
+								break
+							}
+						}
+						if !found {
+							row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
+							colIdx++
+						}
+					}
+				case reflect.Slice:
+					dynVals := reflect.ValueOf(cv)
+					for _, dyncol := range dynamicColumns[col.Name] {
+						found := false
+						for j := 0; j < dynVals.Len(); j++ {
+							pair := reflect.ValueOf(dynVals.Index(j).Interface())
+							if ToSnakeCase(fmt.Sprintf("%v", pair.Field(0).Interface())) == dyncol {
+								row = append(row, parquet.ValueOf(pair.Field(1).Interface()).Level(0, 1, colIdx))
+								colIdx++
+								found = true
+								break
+							}
+						}
+						if !found {
+							row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
+							colIdx++
+						}
+					}
+				case reflect.Map:
+					dynVals := reflect.ValueOf(cv)
+					for _, key := range dynVals.MapKeys() {
+						switch key.Kind() {
+						case reflect.String:
+							for _, dyncol := range dynamicColumns[col.Name] {
+								found := false
+								for _, key := range dynVals.MapKeys() {
+									if ToSnakeCase(fmt.Sprintf("%v", key.String())) == dyncol {
+										row = append(row, parquet.ValueOf(dynVals.MapIndex(key).Interface()).Level(0, 1, colIdx))
+										colIdx++
+										found = true
+										break
+									}
+								}
+
+								if !found {
+									row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
+									colIdx++
+								}
+							}
+						default:
+							return nil, fmt.Errorf("unsupported dynamic map key type for column '%s'", col.Name)
+						}
+					}
+				default:
+					return nil, fmt.Errorf("unsupported dynamic type for column '%s'", col.Name)
+				}
+			default:
+				switch t := cv.(type) {
+				case []uuid.UUID: // Special handling for this type
+					row = append(row, parquet.ValueOf(ExtractLocationIDs(t)).Level(0, 0, colIdx))
+				default:
+					row = append(row, parquet.ValueOf(cv).Level(0, 0, colIdx))
+				}
+				colIdx++
+			}
+		}
+
+		rows = append(rows, row)
+	}
+
+	pb, err := schema.NewBuffer(dynamicColumns)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = pb.WriteRows(rows)
+	if err != nil {
+		return nil, err
+	}
+
+	return pb, nil
+}
+
 // NewBuffer returns a new buffer with a concrete parquet schema generated
 // using the given concrete dynamic column names.
 func (s *Schema) NewBuffer(dynamicColumns map[string][]string) (*Buffer, error) {
@@ -1217,6 +1390,10 @@ func (r *MergedRowGroup) DynamicRows() DynamicRowReader {
 
 type mergeOption struct {
 	dynamicColumns map[string][]string
+	// alreadySorted indicates that the row groups are already sorted and
+	// non-overlapping. This results in a parquet.MultiRowGroup, which is just
+	// a wrapper without the full-scale merging infrastructure.
+	alreadySorted bool
 }
 
 type MergeOption func(m *mergeOption)
@@ -1227,6 +1404,12 @@ func WithDynamicCols(cols map[string][]string) MergeOption {
 	}
 }
 
+func WithAlreadySorted() MergeOption {
+	return func(m *mergeOption) {
+		m.alreadySorted = true
+	}
+}
+
 // MergeDynamicRowGroups merges the given dynamic row groups into a single
 // dynamic row group. It merges the parquet schema in a non-conflicting way by
 // merging all the concrete dynamic column names and generating a superset
@@ -1264,11 +1447,15 @@ func (s *Schema) MergeDynamicRowGroups(rowGroups []DynamicRowGroup, options ...M
 		))
 	}
 
+	var opts []parquet.RowGroupOption
+	if !m.alreadySorted {
+		opts = append(opts, parquet.SortingRowGroupConfig(
+			parquet.SortingColumns(cols...),
+		))
+	}
 	merge, err := parquet.MergeRowGroups(
 		adapters,
-		parquet.SortingRowGroupConfig(
-			parquet.SortingColumns(cols...),
-		),
+		opts...,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create merge row groups: %w", err)
diff --git a/parts/part.go b/parts/part.go
index 029bba5..1961570 100644
--- a/parts/part.go
+++ b/parts/part.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"fmt"
 	"math"
+	"sort"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/parquet-go/parquet-go"
@@ -268,7 +269,7 @@ func NewPartSorter(schema *dynparquet.Schema, parts []*Part) *PartSorter {
 	}
 }
 
-func (p PartSorter) Len() int {
+func (p *PartSorter) Len() int {
 	return len(p.parts)
 }
 
@@ -293,3 +294,63 @@ func (p *PartSorter) Swap(i, j int) {
 func (p *PartSorter) Err() error {
 	return p.err
 }
+
+// FindMaximumNonOverlappingSet removes the minimum number of parts from the
+// given slice in order to return the maximum non-overlapping set of parts.
+// The function returns the non-overlapping parts first and any overlapping
+// parts second. The parts returned are in sorted order according to their Least
+// row.
+func FindMaximumNonOverlappingSet(schema *dynparquet.Schema, parts []*Part) ([]*Part, []*Part, error) {
+	if len(parts) < 2 {
+		return parts, nil, nil
+	}
+	sorter := NewPartSorter(schema, parts)
+	sort.Sort(sorter)
+	if sorter.Err() != nil {
+		return nil, nil, sorter.Err()
+	}
+
+	// Parts are now sorted according to their Least row.
+	end, err := parts[0].most()
+	if err != nil {
+		return nil, nil, err
+	}
+	nonOverlapping := make([]*Part, 0, len(parts))
+	overlapping := make([]*Part, 0, len(parts))
+	for i := 1; i < len(parts); i++ {
+		start, err := parts[i].Least()
+		if err != nil {
+			return nil, nil, err
+		}
+		curEnd, err := parts[i].most()
+		if err != nil {
+			return nil, nil, err
+		}
+		if schema.Cmp(end, start) < 0 {
+			// No overlap, append the previous part and update end for the next
+			// iteration.
+			nonOverlapping = append(nonOverlapping, parts[i-1])
+			end = curEnd
+			continue
+		}
+
+		// This part overlaps with the previous part. Remove the part with
+		// the highest end row.
+		if schema.Cmp(end, curEnd) >= 0 {
+			overlapping = append(overlapping, parts[i-1])
+			end = curEnd
+		} else {
+			// The current part must be removed. Don't update end, this will
+			// be used in the next iteration and must stay the same.
+			overlapping = append(overlapping, parts[i])
+		}
+	}
+	if len(overlapping) == 0 || overlapping[len(overlapping)-1] != parts[len(parts)-1] {
+		// The last part either did not overlap with its previous part, or
+		// overlapped but had a smaller end row than its previous part (so the
+		// previous part is in the overlapping slice). The last part must be
+		// appended to nonOverlapping.
+		nonOverlapping = append(nonOverlapping, parts[len(parts)-1])
+	}
+	return nonOverlapping, overlapping, nil
+}
diff --git a/table.go b/table.go
index 4e33be3..00f1a80 100644
--- a/table.go
+++ b/table.go
@@ -9,11 +9,8 @@ import (
 	"math"
 	"math/rand"
 	"path/filepath"
-	"reflect"
-	"regexp"
 	"runtime"
 	"sort"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -26,7 +23,6 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/google/btree"
-	"github.com/google/uuid"
 	"github.com/oklog/ulid"
 	"github.com/parquet-go/parquet-go"
 	"github.com/prometheus/client_golang/prometheus"
@@ -569,7 +565,7 @@ func (t *Table) EnsureCompaction() error {
 
 // Write objects into the table.
 func (t *Table) Write(ctx context.Context, vals ...any) (uint64, error) {
-	b, err := ValuesToBuffer(t.Schema(), vals...)
+	b, err := dynparquet.ValuesToBuffer(t.Schema(), vals...)
 	if err != nil {
 		return 0, err
 	}
@@ -577,175 +573,6 @@ func (t *Table) Write(ctx context.Context, vals ...any) (uint64, error) {
 	return t.InsertBuffer(ctx, b)
 }
 
-var (
-	matchFirstCap = regexp.MustCompile("(.)([A-Z][a-z]+)")
-	matchAllCap   = regexp.MustCompile("([a-z0-9])([A-Z])")
-)
-
-func ToSnakeCase(str string) string {
-	snake := matchFirstCap.ReplaceAllString(str, "${1}_${2}")
-	snake = matchAllCap.ReplaceAllString(snake, "${1}_${2}")
-	return strings.ToLower(snake)
-}
-
-func ValuesToBuffer(schema *dynparquet.Schema, vals ...any) (*dynparquet.Buffer, error) {
-	dynamicColumns := map[string][]string{}
-	rows := make([]parquet.Row, 0, len(vals))
-
-	findColumn := func(val reflect.Value, col string, v any) any {
-		for i := 0; i < val.NumField(); i++ {
-			if ToSnakeCase(val.Type().Field(i).Name) == col {
-				return val.Field(i).Interface()
-			}
-		}
-		return nil
-	}
-
-	// Collect dynamic columns
-	for _, v := range vals {
-		val := reflect.ValueOf(v)
-		for _, col := range schema.Columns() {
-			cv := findColumn(val, col.Name, v)
-			if cv == nil {
-				continue
-			}
-			switch col.Dynamic {
-			case true:
-				switch reflect.TypeOf(cv).Kind() {
-				case reflect.Struct:
-					dynVals := reflect.ValueOf(cv)
-					for j := 0; j < dynVals.NumField(); j++ {
-						dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(dynVals.Type().Field(j).Name))
-					}
-				case reflect.Slice:
-					dynVals := reflect.ValueOf(cv)
-					for j := 0; j < dynVals.Len(); j++ {
-						pair := reflect.ValueOf(dynVals.Index(j).Interface())
-						dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(fmt.Sprintf("%v", pair.Field(0))))
-					}
-				case reflect.Map:
-					dynVals := reflect.ValueOf(cv)
-					for _, key := range dynVals.MapKeys() {
-						switch key.Kind() {
-						case reflect.String:
-							dynamicColumns[col.Name] = append(dynamicColumns[col.Name], ToSnakeCase(fmt.Sprintf("%v", key.String())))
-						default:
-							return nil, fmt.Errorf("unsupported dynamic map key type for column for column '%s'", col.Name)
-						}
-					}
-				default:
-					return nil, fmt.Errorf("unsupported dynamic type for column '%s'", col.Name)
-				}
-			}
-		}
-	}
-
-	dynamicColumns = bufutils.Dedupe(dynamicColumns)
-
-	// Create all rows
-	for _, v := range vals {
-		row := []parquet.Value{}
-		val := reflect.ValueOf(v)
-
-		colIdx := 0
-		for _, col := range schema.Columns() {
-			cv := findColumn(val, col.Name, v)
-			if cv == nil {
-				continue
-			}
-			switch col.Dynamic {
-			case true:
-				switch reflect.TypeOf(cv).Kind() {
-				case reflect.Struct:
-					dynVals := reflect.ValueOf(cv)
-					for _, dyncol := range dynamicColumns[col.Name] {
-						found := false
-						for j := 0; j < dynVals.NumField(); j++ {
-							if ToSnakeCase(dynVals.Type().Field(j).Name) == dyncol {
-								row = append(row, parquet.ValueOf(dynVals.Field(j).Interface()).Level(0, 1, colIdx))
-								colIdx++
-								found = true
-								break
-							}
-						}
-						if !found {
-							row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
-							colIdx++
-						}
-					}
-				case reflect.Slice:
-					dynVals := reflect.ValueOf(cv)
-					for _, dyncol := range dynamicColumns[col.Name] {
-						found := false
-						for j := 0; j < dynVals.Len(); j++ {
-							pair := reflect.ValueOf(dynVals.Index(j).Interface())
-							if ToSnakeCase(fmt.Sprintf("%v", pair.Field(0).Interface())) == dyncol {
-								row = append(row, parquet.ValueOf(pair.Field(1).Interface()).Level(0, 1, colIdx))
-								colIdx++
-								found = true
-								break
-							}
-						}
-						if !found {
-							row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
-							colIdx++
-						}
-					}
-				case reflect.Map:
-					dynVals := reflect.ValueOf(cv)
-					for _, key := range dynVals.MapKeys() {
-						switch key.Kind() {
-						case reflect.String:
-							for _, dyncol := range dynamicColumns[col.Name] {
-								found := false
-								for _, key := range dynVals.MapKeys() {
-									if ToSnakeCase(fmt.Sprintf("%v", key.String())) == dyncol {
-										row = append(row, parquet.ValueOf(dynVals.MapIndex(key).Interface()).Level(0, 1, colIdx))
-										colIdx++
-										found = true
-										break
-									}
-								}
-
-								if !found {
-									row = append(row, parquet.ValueOf(nil).Level(0, 0, colIdx))
-									colIdx++
-								}
-							}
-						default:
-							return nil, fmt.Errorf("unsupported dynamic map key type for column '%s'", col.Name)
-						}
-					}
-				default:
-					return nil, fmt.Errorf("unsupported dynamic type for column '%s'", col.Name)
-				}
-			default:
-				switch t := cv.(type) {
-				case []uuid.UUID: // Special handling for this type
-					row = append(row, parquet.ValueOf(dynparquet.ExtractLocationIDs(t)).Level(0, 0, colIdx))
-				default:
-					row = append(row, parquet.ValueOf(cv).Level(0, 0, colIdx))
-				}
-				colIdx++
-			}
-		}
-
-		rows = append(rows, row)
-	}
-
-	pb, err := schema.NewBuffer(dynamicColumns)
-	if err != nil {
-		return nil, err
-	}
-
-	_, err = pb.WriteRows(rows)
-	if err != nil {
-		return nil, err
-	}
-
-	return pb, nil
-}
-
 func (t *Table) InsertRecord(ctx context.Context, record arrow.Record) (uint64, error) {
 	block, close, err := t.appender(ctx)
 	if err != nil {


Test patch:
diff --git a/parts/part_test.go b/parts/part_test.go
new file mode 100644
index 0000000..fdd3f93
--- /dev/null
+++ b/parts/part_test.go
@@ -0,0 +1,95 @@
+package parts
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/polarsignals/frostdb/dynparquet"
+	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
+)
+
+func TestFindMaximumNonOverlappingSet(t *testing.T) {
+	testSchema, err := dynparquet.SchemaFromDefinition(&schemapb.Schema{
+		Name: "test_schema",
+		Columns: []*schemapb.Column{{
+			Name: "ints",
+			StorageLayout: &schemapb.StorageLayout{
+				Type:     schemapb.StorageLayout_TYPE_INT64,
+				Encoding: schemapb.StorageLayout_ENCODING_PLAIN_UNSPECIFIED,
+			},
+		}},
+		SortingColumns: []*schemapb.SortingColumn{{Name: "ints", Direction: schemapb.SortingColumn_DIRECTION_ASCENDING}},
+	})
+	require.NoError(t, err)
+
+	type rng struct {
+		start int64
+		end   int64
+	}
+	type dataModel struct {
+		Ints int64
+	}
+	for _, tc := range []struct {
+		name                   string
+		ranges                 []rng
+		expectedNonOverlapping []rng
+		expectedOverlapping    []rng
+	}{
+		{
+			name:                   "SinglePart",
+			ranges:                 []rng{{1, 2}},
+			expectedNonOverlapping: []rng{{1, 2}},
+		},
+		{
+			name:                   "TwoNonOverlapping",
+			ranges:                 []rng{{1, 2}, {3, 4}},
+			expectedNonOverlapping: []rng{{1, 2}, {3, 4}},
+		},
+		{
+			name:                   "OneOverlap",
+			ranges:                 []rng{{1, 2}, {4, 7}, {3, 8}},
+			expectedNonOverlapping: []rng{{1, 2}, {4, 7}},
+			expectedOverlapping:    []rng{{3, 8}},
+		},
+		{
+			name:                   "ChooseMinimumNumber",
+			ranges:                 []rng{{1, 2}, {4, 10}, {4, 5}, {6, 7}},
+			expectedNonOverlapping: []rng{{1, 2}, {4, 5}, {6, 7}},
+			expectedOverlapping:    []rng{{4, 10}},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			parts := make([]*Part, len(tc.ranges))
+			for i := range parts {
+				start := dataModel{Ints: tc.ranges[i].start}
+				end := dataModel{Ints: tc.ranges[i].end}
+				buf, err := dynparquet.ValuesToBuffer(testSchema, start, end)
+				require.NoError(t, err)
+				var b bytes.Buffer
+				require.NoError(t, testSchema.SerializeBuffer(&b, buf))
+				serBuf, err := dynparquet.ReaderFromBytes(b.Bytes())
+				require.NoError(t, err)
+				parts[i] = NewPart(0, serBuf)
+			}
+			nonOverlapping, overlapping, err := FindMaximumNonOverlappingSet(testSchema, parts)
+			require.NoError(t, err)
+
+			verify := func(t *testing.T, expected []rng, actual []*Part) {
+				t.Helper()
+				require.Len(t, actual, len(expected))
+				for i := range actual {
+					start, err := actual[i].Least()
+					require.NoError(t, err)
+					end, err := actual[i].most()
+					require.NoError(t, err)
+					require.Equal(t, expected[i].start, start.Row[0].Int64())
+					require.Equal(t, expected[i].end, end.Row[0].Int64())
+				}
+			}
+			verify(t, tc.expectedNonOverlapping, nonOverlapping)
+			verify(t, tc.expectedOverlapping, overlapping)
+		})
+	}
+}