Repository: polarsignals/frostdb
Pull request #536 (2023-09-19 15:38:22)

Revert "compaction: treat non-overlapping l0 parts as a single row group"
Reverts polarsignals/frostdb#535


Patch:
diff --git a/compaction.go b/compaction.go
index 47cd579..5cac9e4 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 := make([]*parts.Part, 0, len(level1Parts))
+	partsToCompact := level0Parts
 	// 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,8 +542,6 @@ 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 {
@@ -561,38 +559,7 @@ 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 {
@@ -605,6 +572,7 @@ func compactLevel0IntoLevel1(
 		bufs = append(bufs, buf.MultiDynamicRowGroup())
 	}
 
+	cursor := 0
 	merged, err := t.table.schema.MergeDynamicRowGroups(bufs)
 	if err != nil {
 		return nil, err
@@ -641,6 +609,7 @@ 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 53e8ffa..2641c30 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -5,21 +5,17 @@ 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"
 )
@@ -955,175 +951,6 @@ 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) {
@@ -1390,10 +1217,6 @@ 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)
@@ -1404,12 +1227,6 @@ 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
@@ -1447,15 +1264,11 @@ 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,
-		opts...,
+		parquet.SortingRowGroupConfig(
+			parquet.SortingColumns(cols...),
+		),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("create merge row groups: %w", err)
diff --git a/parts/part.go b/parts/part.go
index 1961570..029bba5 100644
--- a/parts/part.go
+++ b/parts/part.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"math"
-	"sort"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/parquet-go/parquet-go"
@@ -269,7 +268,7 @@ func NewPartSorter(schema *dynparquet.Schema, parts []*Part) *PartSorter {
 	}
 }
 
-func (p *PartSorter) Len() int {
+func (p PartSorter) Len() int {
 	return len(p.parts)
 }
 
@@ -294,63 +293,3 @@ 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 00f1a80..4e33be3 100644
--- a/table.go
+++ b/table.go
@@ -9,8 +9,11 @@ import (
 	"math"
 	"math/rand"
 	"path/filepath"
+	"reflect"
+	"regexp"
 	"runtime"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -23,6 +26,7 @@ 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"
@@ -565,7 +569,7 @@ func (t *Table) EnsureCompaction() error {
 
 // Write objects into the table.
 func (t *Table) Write(ctx context.Context, vals ...any) (uint64, error) {
-	b, err := dynparquet.ValuesToBuffer(t.Schema(), vals...)
+	b, err := ValuesToBuffer(t.Schema(), vals...)
 	if err != nil {
 		return 0, err
 	}
@@ -573,6 +577,175 @@ 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
deleted file mode 100644
index fdd3f93..0000000
--- a/parts/part_test.go
+++ /dev/null
@@ -1,95 +0,0 @@
-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)
-		})
-	}
-}