Repository: polarsignals/frostdb
Pull request #550 (2023-09-28 21:07:34)

Lsm like index
This adds the LSM-like index. ( As described in the ![design](https://github.com/polarsignals/frostdb/blob/main/design/lsm.md) this isn't an actual LSM, but one that borrows the idea of merging levels into larger sorted/compacted levels) 

It is implemented as a single linked list with sentinel nodes in the list indicating the start of a new level of the LSM (L0->L1 etc.).  The LSM compacts from L0 into L1 when the size of L0 reaches the configured size, this compaction can cause a cascade where compacting L0 into L1 causes L1 to also require a compaction. Compaction is the least polished part of this implementation thus far, but I wanted to get this our for a round of reviews. 

Below are some benchmarks for this index:

```
thor@thors-MacBook-Pro frostdb % benchstat BenchmarkInsertBTree BenchmarkInsertLSM
name                                          old time/op    new time/op    delta
_Table_Insert_1000Rows_10Iters_10Writers-10      211ms ± 5%      11ms ± 1%  -94.61%  (p=0.000 n=10+10)
_Table_Insert_100Rows_1000Iters_1Writers-10      710ms ± 6%      21ms ± 4%  -96.97%  (p=0.000 n=10+10)
_Table_Insert_100Rows_100Iters_100Writers-10     1.66s ± 5%     0.10s ±12%  -93.88%  (p=0.000 n=9+10)

name                                          old alloc/op   new alloc/op   delta
_Table_Insert_1000Rows_10Iters_10Writers-10      414MB ± 1%       3MB ± 7%  -99.28%  (p=0.000 n=10+10)
_Table_Insert_100Rows_1000Iters_1Writers-10      452MB ± 0%      23MB ± 2%  -94.83%  (p=0.000 n=10+10)
_Table_Insert_100Rows_100Iters_100Writers-10    4.63GB ± 1%    0.30GB ± 6%  -93.62%  (p=0.000 n=10+10)

name                                          old allocs/op  new allocs/op  delta
_Table_Insert_1000Rows_10Iters_10Writers-10      6.59M ± 1%     0.03M ± 0%  -99.61%  (p=0.000 n=10+10)
_Table_Insert_100Rows_1000Iters_1Writers-10      7.17M ± 0%     0.25M ± 0%  -96.49%  (p=0.000 n=9+10)
_Table_Insert_100Rows_100Iters_100Writers-10     72.9M ± 0%      2.5M ± 0%  -96.53%  (p=0.000 n=7+10)
```

```
thor@thors-MacBook-Pro frostdb % benchstat BenchmarkAggregationBTree BenchmarkAggregationLSM
name                  old time/op    new time/op    delta
Aggregation/sum-10       408µs ± 3%     409µs ± 6%    ~     (p=0.853 n=10+10)
Aggregation/count-10     412µs ± 5%     402µs ± 3%  -2.45%  (p=0.008 n=9+10)
Aggregation/max-10       408µs ± 3%     408µs ± 2%    ~     (p=0.579 n=10+10)

name                  old alloc/op   new alloc/op   delta
Aggregation/sum-10       417kB ± 0%     417kB ± 0%    ~     (p=0.631 n=10+10)
Aggregation/count-10     417kB ± 0%     416kB ± 0%  -0.26%  (p=0.003 n=10+10)
Aggregation/max-10       416kB ± 0%     416kB ± 0%  +0.11%  (p=0.017 n=9+10)

name                  old allocs/op  new allocs/op  delta
Aggregation/sum-10         708 ± 0%       704 ± 0%  -0.49%  (p=0.000 n=10+9)
Aggregation/count-10       708 ± 0%       704 ± 0%  -0.51%  (p=0.000 n=10+10)
Aggregation/max-10         707 ± 0%       704 ± 0%  -0.42%  (p=0.000 n=9+10)
```

asubiotto: Overall this is looking very good. The thing I'm most concerned about is regressions related to compactions. I know you've deployed this using parca, but I wonder if we could also figure out a way to deploy this on a production instance?

I also think it would be nice to compare queries on actual production data using `BenchmarkQuery` in `bench_test.go`. I would go about it by using e.g. the prod data folder I sent you and skipping over the part loading from snapshots which would result in just constructing each index version using the inserts from the WAL (followed by a compaction).
asubiotto: I tried running `BenchmarkQuery` on some wal-only prod data with this PR but get a panic when ensuring a compaction:
```
go test -run ^# -bench BenchmarkQuery -count 6 -v | tee benchlsm
goos: darwin
goarch: arm64
pkg: github.com/polarsignals/frostdb
BenchmarkQuery
    bench_test.go:225: recovering DB
    bench_test.go:225: recovered DB in 3.739401042s
    bench_test.go:225: ensuring compaction
panic: runtime error: index out of range [0] with length 0

goroutine 8 [running]:
github.com/apache/arrow/go/v14/arrow/array.(*simpleRecord).Column(0x140001dcc00?, 0x1401ade3328?)
	/Users/asubiotto/go/pkg/mod/github.com/apache/arrow/go/v14@v14.0.0-20230829105931-29819425f428/arrow/array/record.go:234 +0x54
github.com/polarsignals/frostdb/pqarrow.getRecordRow(0x141163fb470?, 0x14006d64450?, {0x1031d5840, 0x14006d64450}, 0x1405e30d818?, {0x14102cc6900, 0x17, 0x102aa8f04?}, {0x140ec017800, 0x17, ...})
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/pqarrow/parquet.go:81 +0xe4
github.com/polarsignals/frostdb/pqarrow.RecordToFile(0x140001dcc00, 0x14114a4d920, {0x1031d5840, 0x14006d64450})
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/pqarrow/parquet.go:142 +0x21c
github.com/polarsignals/frostdb/parts.(*Part).SerializeBuffer(0x140001dcc00?, 0x1031c5638?, 0x140c8251ec0?)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/parts/part.go:54 +0x34
github.com/polarsignals/frostdb/parts.(*Part).AsSerializedBuffer(0x14015d1e8c0, 0x140001dcc00)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/parts/part.go:70 +0xbc
github.com/polarsignals/frostdb.(*Table).compactParts(0x1400052a6e0, {0x1031c5638, 0x140459753b0}, {0x140b46d2000, 0xb6f, 0x8d2c?})
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:1116 +0xc8
github.com/polarsignals/frostdb.(*Table).parquetCompaction(0x8?, {0x140b46d2000, 0xb6f, 0xd50})
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:1086 +0x58
github.com/polarsignals/frostdb.(*LSM).merge(0x140001dd780, 0x0, 0x0)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/lsm.go:296 +0x22c
github.com/polarsignals/frostdb.(*TableBlock).EnsureCompaction(...)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:848
github.com/polarsignals/frostdb.(*Table).EnsureCompaction(0x140000de1c0?)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:518 +0x2c
github.com/polarsignals/frostdb.newDBForBenchmarks({0x1031cb9f0, 0x1038ee3e0}, {0x1031d9870, 0x1400052c000})
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/bench_test.go:62 +0x1dc
github.com/polarsignals/frostdb.BenchmarkQuery(0x1400052c000)
	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/bench_test.go:225 +0x4c
testing.(*B).runN(0x1400052c000, 0x1)
	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:193 +0x128
testing.(*B).run1.func1()
	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:233 +0x50
created by testing.(*B).run1 in goroutine 1
	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:226 +0x90
exit status 2
FAIL	github.com/polarsignals/frostdb	48.054s
```
thorfour: > I tried running `BenchmarkQuery` on some wal-only prod data with this PR but get a panic when ensuring a compaction:
> 
> ```
> go test -run ^# -bench BenchmarkQuery -count 6 -v | tee benchlsm
> goos: darwin
> goarch: arm64
> pkg: github.com/polarsignals/frostdb
> BenchmarkQuery
>     bench_test.go:225: recovering DB
>     bench_test.go:225: recovered DB in 3.739401042s
>     bench_test.go:225: ensuring compaction
> panic: runtime error: index out of range [0] with length 0
> 
> goroutine 8 [running]:
> github.com/apache/arrow/go/v14/arrow/array.(*simpleRecord).Column(0x140001dcc00?, 0x1401ade3328?)
> 	/Users/asubiotto/go/pkg/mod/github.com/apache/arrow/go/v14@v14.0.0-20230829105931-29819425f428/arrow/array/record.go:234 +0x54
> github.com/polarsignals/frostdb/pqarrow.getRecordRow(0x141163fb470?, 0x14006d64450?, {0x1031d5840, 0x14006d64450}, 0x1405e30d818?, {0x14102cc6900, 0x17, 0x102aa8f04?}, {0x140ec017800, 0x17, ...})
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/pqarrow/parquet.go:81 +0xe4
> github.com/polarsignals/frostdb/pqarrow.RecordToFile(0x140001dcc00, 0x14114a4d920, {0x1031d5840, 0x14006d64450})
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/pqarrow/parquet.go:142 +0x21c
> github.com/polarsignals/frostdb/parts.(*Part).SerializeBuffer(0x140001dcc00?, 0x1031c5638?, 0x140c8251ec0?)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/parts/part.go:54 +0x34
> github.com/polarsignals/frostdb/parts.(*Part).AsSerializedBuffer(0x14015d1e8c0, 0x140001dcc00)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/parts/part.go:70 +0xbc
> github.com/polarsignals/frostdb.(*Table).compactParts(0x1400052a6e0, {0x1031c5638, 0x140459753b0}, {0x140b46d2000, 0xb6f, 0x8d2c?})
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:1116 +0xc8
> github.com/polarsignals/frostdb.(*Table).parquetCompaction(0x8?, {0x140b46d2000, 0xb6f, 0xd50})
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:1086 +0x58
> github.com/polarsignals/frostdb.(*LSM).merge(0x140001dd780, 0x0, 0x0)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/lsm.go:296 +0x22c
> github.com/polarsignals/frostdb.(*TableBlock).EnsureCompaction(...)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:848
> github.com/polarsignals/frostdb.(*Table).EnsureCompaction(0x140000de1c0?)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/table.go:518 +0x2c
> github.com/polarsignals/frostdb.newDBForBenchmarks({0x1031cb9f0, 0x1038ee3e0}, {0x1031d9870, 0x1400052c000})
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/bench_test.go:62 +0x1dc
> github.com/polarsignals/frostdb.BenchmarkQuery(0x1400052c000)
> 	/Users/asubiotto/Developer/github.com/polarsignals/frostdb/bench_test.go:225 +0x4c
> testing.(*B).runN(0x1400052c000, 0x1)
> 	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:193 +0x128
> testing.(*B).run1.func1()
> 	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:233 +0x50
> created by testing.(*B).run1 in goroutine 1
> 	/opt/homebrew/opt/go/libexec/src/testing/benchmark.go:226 +0x90
> exit status 2
> FAIL	github.com/polarsignals/frostdb	48.054s
> ```

I neglected to respond to this apologies. But this is fixed now

Patch:
diff --git a/compaction.go b/compaction.go
deleted file mode 100644
index 97bb3e6..0000000
--- a/compaction.go
+++ /dev/null
@@ -1,748 +0,0 @@
-package frostdb
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"math"
-	"runtime"
-	"sort"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/dustin/go-humanize"
-	"github.com/go-kit/log"
-	"github.com/go-kit/log/level"
-	"github.com/google/btree"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/thanos-io/objstore/errutil"
-
-	"github.com/polarsignals/frostdb/dynparquet"
-	"github.com/polarsignals/frostdb/parts"
-)
-
-type CompactionOption func(*CompactionConfig)
-
-type CompactionConfig struct {
-	concurrency          int
-	interval             time.Duration
-	l1ToGranuleSizeRatio float64
-
-	// compactAfterRecovery specifies to run compaction on all tables after recovery.
-	compactAfterRecovery           bool
-	compactAfterRecoveryTableNames []string
-}
-
-// NewCompactionConfig creates a new compaction config with the given options.
-// If none are specified, a default compaction config is created.
-func NewCompactionConfig(options ...CompactionOption) *CompactionConfig {
-	c := &CompactionConfig{
-		concurrency: runtime.GOMAXPROCS(0),
-		interval:    100 * time.Millisecond,
-		// 0.5 was chosen so that a level1 part takes up around half the space
-		// in a full granule, allowing for level0 parts to grow up to the
-		// remaining space before compaction. The higher this ratio, the less
-		// new data is compacted and the more compactions occur. However, a
-		// lower ratio implies leaving some parquet compression size savings on
-		// the table.
-		l1ToGranuleSizeRatio: 0.5,
-	}
-	for _, o := range options {
-		o(c)
-	}
-	return c
-}
-
-// WithCompactionAfterRecovery specifies to run compaction on a specific set
-// of tables after recovery. If a zero-length slice is passed in, compaction is
-// run on all tables (this can be slow since it is currently single-threaded).
-func WithCompactionAfterRecovery(tableNames []string) CompactionOption {
-	return func(c *CompactionConfig) {
-		c.compactAfterRecovery = true
-		c.compactAfterRecoveryTableNames = tableNames
-	}
-}
-
-// WithConcurrency specifies the number of concurrent goroutines compacting data
-// for each database.
-func WithConcurrency(concurrency int) CompactionOption {
-	return func(c *CompactionConfig) {
-		c.concurrency = concurrency
-	}
-}
-
-// WithInterval specifies the compaction sweep interval.
-func WithInterval(i time.Duration) CompactionOption {
-	return func(c *CompactionConfig) {
-		c.interval = i
-	}
-}
-
-// WithL1ToGranuleSizeRatio sets the target level1 part size relative to the
-// granule size. The closer this value is to 1, the more compacted data becomes
-// with an expected rise in memory and CPU usage.
-func WithL1ToGranuleSizeRatio(r float64) CompactionOption {
-	return func(c *CompactionConfig) {
-		c.l1ToGranuleSizeRatio = r
-	}
-}
-
-type compactorPool struct {
-	db *DB
-
-	cfg    *CompactionConfig
-	wg     sync.WaitGroup
-	cancel context.CancelFunc
-	paused atomic.Bool
-}
-
-func newCompactorPool(db *DB, cfg *CompactionConfig) *compactorPool {
-	return &compactorPool{
-		db:  db,
-		cfg: cfg,
-	}
-}
-
-func (c *compactorPool) start() {
-	ctx, cancelFn := context.WithCancel(context.Background())
-	c.cancel = cancelFn
-	for i := 0; i < c.cfg.concurrency; i++ {
-		c.wg.Add(1)
-		go func() {
-			defer c.wg.Done()
-			c.compactLoop(ctx)
-		}()
-	}
-}
-
-func (c *compactorPool) pause() {
-	c.paused.Store(true)
-}
-
-func (c *compactorPool) resume() {
-	c.paused.Store(false)
-}
-
-func (c *compactorPool) stop() {
-	if c.cancel == nil {
-		// Pool was not started.
-		return
-	}
-	c.cancel()
-	c.wg.Wait()
-}
-
-func (c *compactorPool) compactLoop(ctx context.Context) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-time.After(c.cfg.interval):
-			if c.paused.Load() {
-				continue
-			}
-			c.db.mtx.RLock()
-			tablesToCompact := make([]*Table, 0, len(c.db.tables))
-			for _, table := range c.db.tables {
-				tablesToCompact = append(tablesToCompact, table)
-			}
-			c.db.mtx.RUnlock()
-
-			for _, table := range tablesToCompact {
-				block, done, err := table.ActiveWriteBlock() // obtain a write block to prevent compaction and persistence from running at the same time
-				if err != nil {
-					continue // errors are for tables closing
-				}
-				if err := block.compact(c.cfg); err != nil {
-					level.Warn(c.db.logger).Log("msg", "compaction failed", "err", err)
-				}
-				done()
-			}
-		}
-	}
-}
-
-func (t *TableBlock) compact(cfg *CompactionConfig) error {
-	var compactionErrors errutil.MultiError
-	index := t.Index()
-	index.Ascend(func(i btree.Item) bool {
-		granuleToCompact := i.(*Granule)
-		if granuleToCompact.metadata.size.Load() < uint64(t.table.db.columnStore.granuleSizeBytes) {
-			// Skip granule since its size is under the target size.
-			return true
-		}
-		if !granuleToCompact.metadata.pruned.CompareAndSwap(0, 1) {
-			// Someone else claimed this granule compaction.
-			return true
-		}
-
-		defer t.table.metrics.compactions.Inc()
-		if successful, err := t.compactGranule(granuleToCompact, cfg); !successful || err != nil {
-			t.abortCompaction(granuleToCompact)
-			if err != nil {
-				compactionErrors.Add(err)
-			}
-		}
-		return true
-	})
-	return compactionErrors.Err()
-}
-
-// compactionStats is a helper struct to collect metrics during compaction.
-type compactionStats struct {
-	level0SizeBefore        int64
-	level0CountBefore       int
-	level0NumRowsBefore     int64
-	level1SizeBefore        int64
-	level1CountBefore       int
-	level1NumRowsBefore     int64
-	numPartsOverlap         int
-	uncompletedTxnPartCount int64
-	level1SizeAfter         uint64
-	level1CountAfter        int
-	splits                  int
-	totalDuration           time.Duration
-}
-
-func (s compactionStats) recordAndLog(m *compactionMetrics, l log.Logger) {
-	m.level0SizeBefore.Observe(float64(s.level0SizeBefore))
-	m.level0CountBefore.Observe(float64(s.level0CountBefore))
-	m.level1SizeBefore.Observe(float64(s.level1SizeBefore))
-	m.level1CountBefore.Observe(float64(s.level1CountBefore))
-	m.numPartsOverlap.Observe(float64(s.numPartsOverlap))
-	m.uncompletedTxnPartCount.Observe(float64(s.uncompletedTxnPartCount))
-	m.level1SizeAfter.Observe(float64(s.level1SizeAfter))
-	m.level1CountAfter.Observe(float64(s.level1CountAfter))
-	m.splits.Observe(float64(s.splits))
-	m.totalDuration.Observe(float64(s.totalDuration.Seconds()))
-
-	level.Debug(l).Log(
-		"msg", "compaction complete",
-		"duration", s.totalDuration,
-		"l0Before", fmt.Sprintf(
-			"[sz=%s,cnt=%d]", humanize.IBytes(uint64(s.level0SizeBefore)), s.level0CountBefore,
-		),
-		"l1Before", fmt.Sprintf(
-			"[sz=%s,cnt=%d]", humanize.IBytes(uint64(s.level1SizeBefore)), s.level1CountBefore,
-		),
-		"l1After", fmt.Sprintf(
-			"[sz=%s,cnt=%d]", humanize.IBytes(uint64(s.level1SizeAfter)), s.level1CountAfter,
-		),
-		"overlaps", s.numPartsOverlap,
-		"splits", s.splits,
-	)
-}
-
-// compactionMetrics are metrics recorded on each successful compaction.
-type compactionMetrics struct {
-	level0SizeBefore        prometheus.Histogram
-	level0CountBefore       prometheus.Histogram
-	level1SizeBefore        prometheus.Histogram
-	level1CountBefore       prometheus.Histogram
-	numPartsOverlap         prometheus.Histogram
-	uncompletedTxnPartCount prometheus.Histogram
-	level1SizeAfter         prometheus.Histogram
-	level1CountAfter        prometheus.Histogram
-	splits                  prometheus.Histogram
-	totalDuration           prometheus.Histogram
-}
-
-func newCompactionMetrics(reg prometheus.Registerer, granuleSize float64) *compactionMetrics {
-	const (
-		twoKiB = 2 << 10
-		// metricResolution is the number of buckets in a histogram for sizes
-		// and times.
-		metricResolution = 25
-	)
-	minSize := float64(twoKiB)
-	if granuleSize < minSize {
-		// This should only happen in tests.
-		minSize = 1
-		granuleSize = twoKiB
-	}
-	sizeBuckets := prometheus.ExponentialBucketsRange(minSize, granuleSize, metricResolution)
-	timeBuckets := prometheus.ExponentialBuckets(0.5, 2, metricResolution)
-	countBuckets := prometheus.ExponentialBuckets(1, 2, 10)
-	return &compactionMetrics{
-		level0SizeBefore: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_0_size_before",
-			Help:    "Total level 0 size when beginning compaction",
-			Buckets: sizeBuckets,
-		}),
-		level0CountBefore: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_0_count_before",
-			Help:    "Number of level 0 parts when beginning compaction",
-			Buckets: countBuckets,
-		}),
-		level1SizeBefore: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_1_size_before",
-			Help:    "Total level 1 size when beginning compaction",
-			Buckets: sizeBuckets,
-		}),
-		level1CountBefore: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_1_count_before",
-			Help:    "Number of level 0 parts when beginning compaction",
-			Buckets: countBuckets,
-		}),
-		numPartsOverlap: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_num_parts_overlap",
-			Help:    "Number of level 1 parts that overlapped with level 0 parts",
-			Buckets: countBuckets,
-		}),
-		uncompletedTxnPartCount: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_uncompleted_txn_part_count",
-			Help:    "Number of parts with uncompleted txns that could not be compacted",
-			Buckets: countBuckets,
-		}),
-		level1SizeAfter: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_1_size_after",
-			Help:    "Total level 1 size after compaction",
-			Buckets: sizeBuckets,
-		}),
-		level1CountAfter: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_level_1_count_after",
-			Help:    "Number of level 1 parts after compaction",
-			Buckets: countBuckets,
-		}),
-		splits: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_splits",
-			Help:    "Number of granule splits",
-			Buckets: countBuckets,
-		}),
-		totalDuration: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-			Name:    "frostdb_compaction_total_duration_seconds",
-			Help:    "Total compaction duration",
-			Buckets: timeBuckets,
-		}),
-	}
-}
-
-// compactGranule is the "meat" of granule compaction. It returns whether the
-// compaction succeeded or not as well as an error (there are cases in which
-// false, nil is returned).
-// The goal of this method is to optimize both memory usage and scan speed.
-// Memory usage is optimized by merging multiple row groups into fewer row
-// groups, since the more data there is in a row group, the higher the benefits
-// of things like parquet compression (this is applied at the row group level).
-// Optimizing for memory usage is a tradeoff, since the more data there is in a
-// row group, the more sequential reads need to be done to find data that
-// queries want. This is why we aim to have a granule maximum size, since
-// searching for granules that satisfy query filters is an
-// O(log(number of granules)) operation, so the more granules we have, the less
-// O(n) work we have to do to search for row groups that satisfy query filters.
-// The compaction logic is as follows:
-//
-//  1. Before any compaction has happened, granules contain a variable number of
-//     parts (basically an encapsulation of a parquet file). These parts contain
-//     a variable number of variable-length row groups. The row group size in
-//     this case is the buffer size that was used by the client on the
-//     corresponding insert call. The compaction strategy in this simple case is
-//     to rewrite the row groups into multiple parts that contain optimally
-//     sized row groups (set by the user using the WithRowGroupSize table
-//     option). The new part sizes aim to be
-//     l1ToGranuleSizeRatio*granuleSize bytes large (specified in the config).
-//     The resulting parts are marked as compacted using compactionLevel1.
-//
-//  2. As more data is added to a granule compacted in point 1, the new data
-//     will again be stored in a suboptimal multi-part format with multiple
-//     variable length row groups. The same compaction as point 1 is performed
-//     but only on the new data as long as it does not overlap with any level1
-//     data (otherwise the overlapping parts are merged into new level1 parts).
-//     If the sum of the part sizes is above the target granule size, a split is
-//     performed, which simply means that the parts will be assigned to new
-//     granules.
-//
-// Some things to be aware of:
-//   - The current compaction strategy assumes there is no benefit to having
-//     muliple parts over a single part as long as the row groups are of equal
-//     length.
-//   - Currently, our compaction strategy is based on a sweep on a timed
-//     interval. This is OK for now, but ideally we would track metrics that
-//     are proxies for memory size/scan speed that would trigger compactions.
-//     For example, the ratio of uncompacted bytes to total bytes. Or performing
-//     split operations based on load.
-//
-// A note on ordering guarantees:
-// FrostDB guarantees ordering amongst granules. This means that if there are
-// two granules in the index, all rows in granule A are less than granule B.
-// However, within a granule, rows are only ordered at a part level since parts
-// are created when a client appends rows and clients are not required to
-// append in order. The current compaction strategy ensures that ordering
-// guarantees between granules are upheld. Additionally, level1 parts will never
-// overlap.
-func (t *TableBlock) compactGranule(granule *Granule, cfg *CompactionConfig) (bool, error) {
-	// Use the latest watermark as the tx id.
-	tx := t.table.db.tx.Load()
-
-	start := time.Now()
-	level0Parts, level1Parts, partsWithUncompletedTxns, stats, err := collectPartsForCompaction(
-		// Start compaction by adding a sentinel node to its parts list.
-		tx, granule.parts.Sentinel(parts.Compacting),
-	)
-	if err != nil {
-		return false, err
-	}
-
-	if len(level0Parts) == 0 && len(level1Parts) == 0 {
-		// Edge case in which only parts with uncompleted txns are found and
-		// they are already bigger than the max size. A future compaction cycle
-		// will take care of this, so return that the compaction was not
-		// successful
-		return false, nil
-	}
-
-	partsBefore := len(level0Parts) + len(level1Parts)
-	if len(level0Parts) > 0 {
-		var err error
-		level1Parts, err = compactLevel0IntoLevel1(
-			t, tx, level0Parts, level1Parts, cfg.l1ToGranuleSizeRatio, &stats,
-		)
-		if err != nil {
-			return false, err
-		}
-		stats.level1CountAfter += len(level1Parts)
-	}
-
-	// This sort is done to simplify the splitting of non-overlapping parts
-	// amongst new granules. There might be something more clever we could do
-	// here if we maintain stronger guarantees about the part ordering given
-	// that level1 parts should already be sorted when collected at the start
-	// of this method (since this is the only place level1 parts get added to
-	// granules). However, it seems like new parts are prepended so the sort
-	// order is inverted. Another option is to recursively split using
-	// addPartToGranule, which would be cheap in the general case.
-	sorter := parts.NewPartSorter(t.table.schema, level1Parts)
-	sort.Sort(sorter)
-	if sorter.Err() != nil {
-		return false, fmt.Errorf("error sorting level1: %w", sorter.Err())
-	}
-
-	newParts, err := divideLevel1PartsForGranule(
-		t,
-		tx,
-		level1Parts,
-		// Set a maximum size of the l1 ratio to give "breathing room" to new
-		// level0 parts. Otherwise, compactions will be initiated more often
-		// and work on less data. This ends up creating granules half filled
-		// up with a level1 part.
-		int64(float64(t.table.db.columnStore.granuleSizeBytes)*cfg.l1ToGranuleSizeRatio),
-	)
-	if err != nil {
-		return false, err
-	}
-	stats.splits = len(newParts) - 1
-
-	newGranules := make([]*Granule, 0, len(newParts))
-	partsAfter := 0
-	for _, parts := range newParts {
-		partsAfter += len(parts)
-		newGranule, err := NewGranule(t.table.schema, parts...)
-		if err != nil {
-			if err != nil {
-				return false, fmt.Errorf("failed to create new granule: %w", err)
-			}
-		}
-		t.table.metrics.granulesCreated.Inc()
-		newGranules = append(newGranules, newGranule)
-	}
-
-	// Size calculation is done before adding the ignored parts to the new
-	// granules.
-	for _, g := range newGranules {
-		stats.level1SizeAfter += g.metadata.size.Load()
-	}
-
-	// Add remaining parts onto new granules.
-	for _, p := range partsWithUncompletedTxns {
-		if err := addPartToGranule(newGranules, p); err != nil {
-			return false, fmt.Errorf("add parts to granules: %w", err)
-		}
-	}
-
-	// We disable compaction for new granules before allowing new inserts to be
-	// propagated to them.
-	for _, childGranule := range newGranules {
-		childGranule.metadata.pruned.Store(1)
-	}
-
-	// We restore the possibility to trigger compaction after we exit the
-	// function.
-	defer func() {
-		for _, childGranule := range newGranules {
-			childGranule.metadata.pruned.Store(0)
-		}
-	}()
-
-	// Set the newGranules pointer, so new writes will propogate into these new
-	// granules.
-	granule.newGranules.Store(&newGranules)
-
-	// Mark compaction complete in the granule; this will cause new writes to
-	// start using the newGranules pointer. Iterate over the resulting part
-	// list to copy any new parts that were added while we were compacting.
-	var addPartErr error
-	granule.parts.Sentinel(parts.Compacted).Iterate(func(p *parts.Part) bool {
-		if err := addPartToGranule(newGranules, p); err != nil {
-			addPartErr = err
-			return false
-		}
-		return true
-	})
-	if addPartErr != nil {
-		return false, fmt.Errorf("add part to granules: %w", addPartErr)
-	}
-
-	if err := t.updateIndex(func(index *btree.BTree) error {
-		if index.Delete(granule) == nil {
-			level.Error(t.logger).Log("msg", "failed to delete granule during split")
-			return fmt.Errorf("failed to delete granule")
-		}
-
-		for _, g := range newGranules {
-			if dupe := index.ReplaceOrInsert(g); dupe != nil {
-				level.Error(t.logger).Log("duplicate insert performed")
-			}
-		}
-
-		return nil
-	}); err != nil {
-		return false, err
-	}
-	sizeDiff := int64(stats.level1SizeAfter) - (stats.level0SizeBefore + stats.level1SizeBefore)
-	t.size.Add(sizeDiff)
-	t.table.metrics.numParts.Add(float64(int(stats.level1CountAfter) - partsBefore))
-
-	stats.totalDuration = time.Since(start)
-	stats.recordAndLog(t.table.metrics.compactionMetrics, t.logger)
-	// Release all records in L0 parts
-	for _, p := range level0Parts {
-		if r := p.Record(); r != nil {
-			r.Release()
-		}
-	}
-	return true, nil
-}
-
-// compactLevel0IntoLevel1 compacts the given level0Parts into level1Parts by
-// merging them with overlapping level1 parts and producing parts that are up
-// to the TableBlock's maximum granule size. These parts are guaranteed to be
-// non-overlapping.
-// The size limit is a best effort as there is currently no easy way to
-// determine the size of a parquet file while writing row groups to it.
-func compactLevel0IntoLevel1(
-	t *TableBlock,
-	tx uint64,
-	level0Parts,
-	level1Parts []*parts.Part,
-	l1ToGranuleSizeRatio float64,
-	stats *compactionStats,
-) ([]*parts.Part, error) {
-	// Verify whether the level0 parts overlap with level1 parts. If they do,
-	// we need to merge the overlapping parts. Not merging these parts could
-	// 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))
-	// 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.
-	size := stats.level1SizeBefore
-	numRows := stats.level1NumRowsBefore
-	if len(level1Parts) == 0 {
-		// If no level1 parts exist, then we estimate the parquet rows to
-		// write based on level0.
-		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 {
-			if overlaps, err := p0.OverlapsWith(t.table.schema, p1); err != nil {
-				return nil, err
-			} else if overlaps {
-				stats.numPartsOverlap++
-				partsToCompact = append(partsToCompact, p1)
-				overlapped = true
-				break
-			}
-		}
-		if !overlapped {
-			nonOverlapping = append(nonOverlapping, p1)
-		}
-	}
-
-	// 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 {
-			return nil, err
-		}
-
-		// All the row groups in a part are wrapped in a single row group given
-		// that all rows are sorted within a part. This reduces the number of
-		// cursors open when merging the row groups.
-		bufs = append(bufs, buf.MultiDynamicRowGroup())
-	}
-
-	merged, err := t.table.schema.MergeDynamicRowGroups(bufs)
-	if err != nil {
-		return nil, err
-	}
-
-	estimatedBytesPerRow := float64(size) / float64(numRows)
-	estimatedRowsPerPart := int(
-		math.Ceil(
-			(float64(t.table.db.columnStore.granuleSizeBytes) * l1ToGranuleSizeRatio) /
-				estimatedBytesPerRow,
-		),
-	)
-	if err := func() error {
-		rows := merged.Rows()
-		defer rows.Close()
-
-		for {
-			var mergedBytes bytes.Buffer
-			n, err := func() (int, error) {
-				writer, err := t.rowWriter(&mergedBytes, merged.DynamicColumns(), withMaxRows(estimatedRowsPerPart))
-				if err != nil {
-					return 0, err
-				}
-				defer writer.close()
-				n, err := writer.writeRows(rows)
-				if err != nil {
-					return 0, err
-				}
-				return n, nil
-			}()
-			if err != nil {
-				return err
-			}
-			if n == 0 {
-				break
-			}
-			serBuf, err := dynparquet.ReaderFromBytes(mergedBytes.Bytes())
-			if err != nil {
-				return err
-			}
-			compactedPart := parts.NewPart(tx, serBuf, parts.WithCompactionLevel(parts.CompactionLevel1))
-			nonOverlapping = append(nonOverlapping, compactedPart)
-		}
-		return nil
-	}(); err != nil {
-		return nil, fmt.Errorf("failed level0->level1 compaction: %w", err)
-	}
-
-	return nonOverlapping, nil
-}
-
-func collectPartsForCompaction(tx uint64, list *parts.List) (
-	level0Parts, level1Parts, partsWithUncompletedTxns []*parts.Part,
-	stats compactionStats, err error,
-) {
-	list.Iterate(func(p *parts.Part) bool {
-		if p.TX() > tx {
-			if !p.HasTombstone() {
-				partsWithUncompletedTxns = append(partsWithUncompletedTxns, p)
-				stats.uncompletedTxnPartCount++
-			}
-			// Parts with a tombstone are dropped.
-			return true
-		}
-
-		switch cl := p.CompactionLevel(); cl {
-		case parts.CompactionLevel0:
-			level0Parts = append(level0Parts, p)
-			stats.level0CountBefore++
-			stats.level0SizeBefore += p.Size()
-			stats.level0NumRowsBefore += p.NumRows()
-		case parts.CompactionLevel1:
-			level1Parts = append(level1Parts, p)
-			stats.level1CountBefore++
-			stats.level1SizeBefore += p.Size()
-			stats.level1NumRowsBefore += p.NumRows()
-		default:
-			err = fmt.Errorf("unexpected part compaction level %d", cl)
-			return false
-		}
-		return true
-	})
-	return
-}
-
-// divideLevel1PartsForGranule returns a two-dimensional slice of parts where
-// each element is a slice of parts that all together have a sum of sizes less
-// than maxSize.
-// The global sort order of the input parts is maintained.
-func divideLevel1PartsForGranule(t *TableBlock, tx uint64, level1 []*parts.Part, maxSize int64) ([][]*parts.Part, error) {
-	var totalSize int64
-	sizes := make([]int64, len(level1))
-	for i, p := range level1 {
-		size := p.Size()
-		totalSize += size
-		sizes[i] = size
-	}
-	if totalSize <= maxSize {
-		// No splits needed.
-		return [][]*parts.Part{level1}, nil
-	}
-
-	// We want to maximize the size of each split slice, so we follow a greedy
-	// approach. Note that in practice because we create level1 parts that are
-	// around half the size of a granule (level1ToGranuleSizeRatio), if a
-	// granule is split, it will be split into two granules that are half filled
-	// up with a level1 part.
-	var (
-		runningSize int64
-		newParts    [][]*parts.Part
-	)
-	for i, size := range sizes {
-		runningSize += size
-		if runningSize > maxSize {
-			// Close the current subslice and create a new one.
-			newParts = append(newParts, []*parts.Part{level1[i]})
-			runningSize = size
-			continue
-		}
-		if i == 0 {
-			newParts = append(newParts, []*parts.Part{level1[i]})
-		} else {
-			newParts[len(newParts)-1] = append(newParts[len(newParts)-1], level1[i])
-		}
-	}
-	return newParts, nil
-}
diff --git a/db.go b/db.go
index 24097b5..35d4b40 100644
--- a/db.go
+++ b/db.go
@@ -17,7 +17,6 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/ipc"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
-	"github.com/google/btree"
 	"github.com/oklog/ulid"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -31,6 +30,7 @@ import (
 	schemav2pb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha2"
 	tablepb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/table/v1alpha1"
 	walpb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/wal/v1alpha1"
+	"github.com/polarsignals/frostdb/index"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 	"github.com/polarsignals/frostdb/storage"
 	"github.com/polarsignals/frostdb/wal"
@@ -47,7 +47,6 @@ type ColumnStore struct {
 	storagePath         string
 	enableWAL           bool
 	manualBlockRotation bool
-	compactionConfig    *CompactionConfig
 	snapshotTriggerSize int64
 	metrics             metrics
 
@@ -55,10 +54,15 @@ type ColumnStore struct {
 	indexDegree int
 	// splitSize is the number of new granules that are created when granules are split (default =2)
 	splitSize int
+	// indexConfig is the configuration settings for the lsm index
+	indexConfig []*index.LevelConfig
 
 	sources []DataSource
 	sinks   []DataSink
 
+	compactAfterRecovery           bool
+	compactAfterRecoveryTableNames []string
+
 	// testingOptions are options only used for testing purposes.
 	testingOptions struct {
 		disableReclaimDiskSpaceOnSnapshot bool
@@ -82,11 +86,11 @@ func New(
 		reg:              prometheus.NewRegistry(),
 		logger:           log.NewNopLogger(),
 		tracer:           trace.NewNoopTracerProvider().Tracer(""),
+		indexConfig:      DefaultIndexConfig,
 		indexDegree:      2,
 		splitSize:        2,
 		granuleSizeBytes: 1 * 1024 * 1024,   // 1MB granule size before splitting
 		activeMemorySize: 512 * 1024 * 1024, // 512MB
-		compactionConfig: NewCompactionConfig(),
 	}
 
 	for _, option := range options {
@@ -213,9 +217,17 @@ func WithStoragePath(path string) Option {
 	}
 }
 
-func WithCompactionConfig(c *CompactionConfig) Option {
+func WithIndexConfig(indexConfig []*index.LevelConfig) Option {
+	return func(s *ColumnStore) error {
+		s.indexConfig = indexConfig
+		return nil
+	}
+}
+
+func WithCompactionAfterRecovery(tableNames []string) Option {
 	return func(s *ColumnStore) error {
-		s.compactionConfig = c
+		s.compactAfterRecovery = true
+		s.compactAfterRecoveryTableNames = tableNames
 		return nil
 	}
 }
@@ -291,11 +303,10 @@ func (s *ColumnStore) recoverDBsFromStorage(ctx context.Context) error {
 		databaseName := f.Name()
 		g.Go(func() error {
 			// Open the DB for the side effect of the snapshot and WALs being loaded as part of the open operation.
-			_, err := s.DB(
-				ctx,
+			_, err := s.DB(ctx,
 				databaseName,
 				WithCompactionAfterOpen(
-					s.compactionConfig.compactAfterRecovery, s.compactionConfig.compactAfterRecoveryTableNames,
+					s.compactAfterRecovery, s.compactAfterRecoveryTableNames,
 				),
 			)
 			return err
@@ -340,8 +351,6 @@ type DB struct {
 	compactAfterRecovery           bool
 	compactAfterRecoveryTableNames []string
 
-	compactorPool *compactorPool
-
 	snapshotInProgress atomic.Bool
 
 	metrics *dbMetrics
@@ -451,7 +460,6 @@ func (s *ColumnStore) DB(ctx context.Context, name string, opts ...DBOption) (*D
 			return err
 		}
 		db.txPool = NewTxPool(&db.highWatermark)
-		db.compactorPool = newCompactorPool(db, s.compactionConfig)
 		// Wait to start the compactor pool since benchmarks show that WAL
 		// replay is a lot more efficient if it is not competing against
 		// compaction. Additionally, if the CompactAfterRecovery option is
@@ -543,7 +551,6 @@ func (s *ColumnStore) DB(ctx context.Context, name string, opts ...DBOption) (*D
 		}
 	}
 
-	db.compactorPool.start()
 	s.dbs[name] = db
 	return db, nil
 }
@@ -641,7 +648,14 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 				// already been persisted.
 				db.mtx.Lock()
 				if table, ok := db.tables[e.TableBlockPersisted.TableName]; ok {
-					table.ActiveBlock().index.Store(btree.New(table.db.columnStore.indexDegree))
+					table.ActiveBlock().index, err = index.NewLSM(
+						table.name,
+						table.configureLSMLevels(db.columnStore.indexConfig),
+						index.LSMWithMetrics(table.metrics.indexMetrics),
+					)
+					if err != nil {
+						return err
+					}
 				}
 				db.mtx.Unlock()
 			}
@@ -898,10 +912,6 @@ func (db *DB) closeInternal() error {
 		db.txPool.Stop()
 	}
 
-	if db.compactorPool != nil {
-		db.compactorPool.stop()
-	}
-
 	return nil
 }
 
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 52f730c..d2fbc83 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -1267,6 +1267,47 @@ type ParquetWriter interface {
 	Reset(writer io.Writer)
 }
 
+func (s *Schema) NewSortingWriter(w io.Writer, dynamicColumns map[string][]string) (ParquetWriter, error) {
+	ps, err := s.GetDynamicParquetSchema(dynamicColumns)
+	if err != nil {
+		return nil, err
+	}
+	defer s.PutPooledParquetSchema(ps)
+
+	cols := s.ParquetSortingColumns(dynamicColumns)
+	bloomFilterColumns := make([]parquet.BloomFilterColumn, 0, len(cols))
+	for _, col := range cols {
+		// Don't add bloom filters to boolean columns
+		colName := strings.Split(col.Path()[0], ".")[0]
+		def, ok := s.ColumnByName(colName)
+		if !ok {
+			continue
+		}
+		if def.StorageLayout.Type().Kind() == parquet.Boolean {
+			continue
+		}
+
+		bloomFilterColumns = append(
+			bloomFilterColumns, parquet.SplitBlockFilter(bloomFilterBitsPerValue, col.Path()...),
+		)
+	}
+
+	return parquet.NewSortingWriter[any](w,
+		32*1024,
+		ps.Schema,
+		parquet.ColumnIndexSizeLimit(ColumnIndexSize),
+		parquet.BloomFilters(bloomFilterColumns...),
+		parquet.KeyValueMetadata(
+			DynamicColumnsKey,
+			serializeDynamicColumns(dynamicColumns),
+		),
+		parquet.SortingWriterConfig(
+			parquet.SortingColumns(cols...),
+			parquet.DropDuplicatedRows(s.uniquePrimaryIndex),
+		),
+	), nil
+}
+
 type PooledWriter struct {
 	pool *sync.Pool
 	ParquetWriter
diff --git a/go.mod b/go.mod
index b0ad5de..419a47c 100644
--- a/go.mod
+++ b/go.mod
@@ -10,11 +10,10 @@ require (
 	github.com/dgryski/go-metro v0.0.0-20211217172704-adc40b04c140
 	github.com/dustin/go-humanize v1.0.1
 	github.com/go-kit/log v0.2.1
-	github.com/google/btree v1.0.1
 	github.com/google/uuid v1.3.0
 	github.com/oklog/ulid v1.3.1
 	github.com/olekukonko/tablewriter v0.0.5
-	github.com/parquet-go/parquet-go v0.0.0-20230717201923-bb12c19a1110
+	github.com/parquet-go/parquet-go v0.18.1-0.20231004061202-cde8189c4c26
 	github.com/pingcap/tidb/parser v0.0.0-20220921115303-5aab87679fde
 	github.com/polarsignals/wal v0.0.0-20230809145250-99ef415c5a6c
 	github.com/prometheus/client_golang v1.16.0
diff --git a/go.sum b/go.sum
index 490e126..17bb93e 100644
--- a/go.sum
+++ b/go.sum
@@ -46,8 +46,6 @@ github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5y
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
 github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
-github.com/google/btree v1.0.1 h1:gK4Kx5IaGY9CD5sPJ36FHiBJ6ZXl0kilRiiCj+jdYp4=
-github.com/google/btree v1.0.1/go.mod h1:xXMiIv4Fb/0kKde4SpL7qlzvu5cMJDRkFDxJfI9uaxA=
 github.com/google/flatbuffers v23.1.21+incompatible h1:bUqzx/MXCDxuS0hRJL2EfjyZL3uQrPbMocUa8zGqsTA=
 github.com/google/flatbuffers v23.1.21+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -76,8 +74,8 @@ github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
 github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
-github.com/parquet-go/parquet-go v0.0.0-20230717201923-bb12c19a1110 h1:f/5EoFhLy7p0JmY3ATEqeET8ZugvjCxas9tJ8GoPkx0=
-github.com/parquet-go/parquet-go v0.0.0-20230717201923-bb12c19a1110/go.mod h1:6pu/Ca02WRyWyF6jbY1KceESGBZMsRMSijjLbajXaG8=
+github.com/parquet-go/parquet-go v0.18.1-0.20231004061202-cde8189c4c26 h1:46gJLpBloArJ3oU2AZpeBQgZmgLzE+iDCOgxz78cwgk=
+github.com/parquet-go/parquet-go v0.18.1-0.20231004061202-cde8189c4c26/go.mod h1:6pu/Ca02WRyWyF6jbY1KceESGBZMsRMSijjLbajXaG8=
 github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
 github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
diff --git a/granule.go b/granule.go
deleted file mode 100644
index 868f0bc..0000000
--- a/granule.go
+++ /dev/null
@@ -1,205 +0,0 @@
-package frostdb
-
-import (
-	"context"
-	"fmt"
-	"sync/atomic"
-
-	"github.com/apache/arrow/go/v14/arrow"
-	"github.com/google/btree"
-
-	"github.com/polarsignals/frostdb/dynparquet"
-	"github.com/polarsignals/frostdb/parts"
-)
-
-type Granule struct {
-	metadata GranuleMetadata
-
-	parts  *parts.List
-	schema *dynparquet.Schema
-
-	// newGranules are the granules that were created after a split
-	newGranules *atomic.Pointer[[]*Granule]
-}
-
-// GranuleMetadata is the metadata for a granule.
-type GranuleMetadata struct {
-	// least is the row that exists within the Granule that is the least.
-	// This is used for quick insertion into the btree, without requiring an
-	// iterator.
-	least *dynparquet.DynamicRow
-
-	// size is the raw commited, and uncommited size of the granule. It is used as a suggestion for potential compaction
-	size atomic.Uint64
-
-	// pruned indicates if this Granule is longer found in the index
-	pruned atomic.Uint64
-}
-
-func NewGranule(schema *dynparquet.Schema, prts ...*parts.Part) (*Granule, error) {
-	g := &Granule{
-		parts:       parts.NewList(&atomic.Pointer[parts.Node]{}, parts.None),
-		schema:      schema,
-		metadata:    GranuleMetadata{},
-		newGranules: &atomic.Pointer[[]*Granule]{},
-	}
-	g.newGranules.Store(&[]*Granule{})
-
-	for _, p := range prts {
-		if err := g.addPart(p); err != nil {
-			return nil, err
-		}
-	}
-
-	return g, nil
-}
-
-func (g *Granule) addPart(p *parts.Part) error {
-	if p.NumRows() == 0 {
-		return nil
-	}
-
-	r, err := p.Least()
-	if err != nil {
-		return err
-	}
-
-	_ = g.parts.Prepend(p)
-	g.metadata.size.Add(uint64(p.Size()))
-
-	if g.metadata.least == nil || g.schema.RowLessThan(r, g.metadata.least) {
-		g.metadata.least = r
-	}
-
-	return nil
-}
-
-// Append adds a part into the Granule. It returns the new size of the Granule.
-func (g *Granule) Append(p *parts.Part) (uint64, error) {
-	node := g.parts.Prepend(p)
-	newSize := g.metadata.size.Add(uint64(p.Size()))
-
-	// If the prepend returned that we're adding to the compacted list; then we
-	// need to propagate the Part to the new granules.
-	if node.Compacted() {
-		err := addPartToGranule(*g.newGranules.Load(), p)
-		if err != nil {
-			return 0, err
-		}
-	}
-
-	return newSize, nil
-}
-
-// PartsForTx returns the parts for the given transaction constraints.
-func (g *Granule) PartsForTx(watermark uint64, iterator func(*parts.Part) bool) {
-	collected := []*parts.Part{}
-	containsRecords := false
-	g.parts.Iterate(func(p *parts.Part) bool {
-		// Don't iterate over parts from an uncompleted transaction
-		if p.TX() > watermark {
-			return true
-		}
-
-		if r := p.Record(); r != nil {
-			containsRecords = true
-			r.Retain()
-		}
-
-		collected = append(collected, p)
-		return true
-	})
-
-	if len(*g.newGranules.Load()) != 0 && containsRecords { // This granule was pruned while we were retaining Records; it's not safe to use them anymore
-		for _, p := range collected {
-			if r := p.Record(); r != nil {
-				r.Release()
-			}
-		}
-		for _, newGran := range *g.newGranules.Load() {
-			newGran.PartsForTx(watermark, iterator)
-		}
-	} else {
-		for _, r := range collected {
-			iterator(r)
-		}
-	}
-}
-
-// Less implements the btree.Item interface.
-func (g *Granule) Less(than btree.Item) bool {
-	var otherRow *dynparquet.DynamicRow
-	switch v := than.(type) {
-	case *Granule:
-		otherRow = v.Least()
-	case btreeComparableDynamicRow:
-		otherRow = v.DynamicRow
-	default:
-		panic(fmt.Sprintf("cannot compare against %T", v))
-	}
-	return g.schema.RowLessThan(g.Least(), otherRow)
-}
-
-// Least returns the least row in a Granule.
-func (g *Granule) Least() *dynparquet.DynamicRow {
-	return g.metadata.least
-}
-
-// Collect will filter row groups or arrow records into the collector. Arrow records passed to the collector must be Released().
-func (g *Granule) Collect(ctx context.Context, tx uint64, filter TrueNegativeFilter, collector chan<- any) {
-	records := []arrow.Record{}
-	g.parts.Iterate(func(p *parts.Part) bool {
-		// Don't iterate over parts from an uncompleted transaction
-		if p.TX() > tx {
-			return true
-		}
-
-		if r := p.Record(); r != nil {
-			r.Retain()
-			records = append(records, r)
-			return true
-		}
-
-		var buf *dynparquet.SerializedBuffer
-		var err error
-		buf, err = p.AsSerializedBuffer(g.schema)
-		if err != nil {
-			return false
-		}
-		f := buf.ParquetFile()
-		for i := range f.RowGroups() {
-			rg := buf.DynamicRowGroup(i)
-			var mayContainUsefulData bool
-			mayContainUsefulData, err = filter.Eval(rg)
-			if err != nil {
-				return false
-			}
-			if mayContainUsefulData {
-				select {
-				case <-ctx.Done():
-					return false
-				case collector <- rg:
-				}
-			}
-		}
-
-		return true
-	})
-
-	if len(*g.newGranules.Load()) != 0 && len(records) != 0 { // This granule was pruned while we were retaining Records; it's not safe to use them anymore
-		for _, r := range records {
-			r.Release()
-		}
-		for _, newGran := range *g.newGranules.Load() {
-			newGran.Collect(ctx, tx, filter, collector)
-		}
-	} else {
-		for _, r := range records {
-			select {
-			case <-ctx.Done():
-				return
-			case collector <- r:
-			}
-		}
-	}
-}
diff --git a/index/lsm.go b/index/lsm.go
new file mode 100644
index 0000000..2ed79eb
--- /dev/null
+++ b/index/lsm.go
@@ -0,0 +1,428 @@
+package index
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/apache/arrow/go/v14/arrow"
+	"github.com/apache/arrow/go/v14/arrow/util"
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/polarsignals/frostdb/dynparquet"
+	"github.com/polarsignals/frostdb/parts"
+	"github.com/polarsignals/frostdb/query/expr"
+	"github.com/polarsignals/frostdb/query/logicalplan"
+)
+
+// LSM is a log-structured merge-tree like index. It is implemented as a single linked list of parts.
+//
+// Arrow records are always added to the L0 list. When a list reaches it's configured max size it is compacted
+// calling the levels Compact function and is then added as a new part to the next level.
+//
+// [L0]->[record]->[record]->[L1]->[record/parquet]->[record/parquet] etc.
+type LSM struct {
+	sync.RWMutex
+	compacting *atomic.Bool
+
+	prefix  string
+	levels  *Node
+	sizes   []atomic.Int64
+	configs []*LevelConfig
+
+	logger  log.Logger
+	metrics *LSMMetrics
+}
+
+// LSMMetrics are the metrics for an LSM index.
+type LSMMetrics struct {
+	Compactions        *prometheus.CounterVec
+	LevelSize          *prometheus.GaugeVec
+	CompactionDuration prometheus.Histogram
+}
+
+// LevelConfig is the configuration for a level in the LSM tree.
+// The Level is the sentinel node that represents the level.
+// The MaxSize is the maximum size in bytes that the level can reach before it triggers compaction into the next level.
+// The Compact function is called when the level reaches it's max size. NOTE: that this is not yet implemtened and the database will rotate the full block as normal.
+type LevelConfig struct {
+	Level   SentinelType
+	MaxSize int64
+	Compact func([]*parts.Part) ([]*parts.Part, int64, int64, error)
+}
+
+type LSMOption func(*LSM)
+
+func LSMWithLogger(logger log.Logger) LSMOption {
+	return func(l *LSM) {
+		l.logger = logger
+	}
+}
+
+func LSMWithMetrics(metrics *LSMMetrics) LSMOption {
+	return func(l *LSM) {
+		l.metrics = metrics
+	}
+}
+
+func NewLSMMetrics(reg prometheus.Registerer) *LSMMetrics {
+	return &LSMMetrics{
+		Compactions: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Name: "frostdb_lsm_compactions_total",
+			Help: "The total number of compactions that have occurred.",
+		}, []string{"level"}),
+
+		LevelSize: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
+			Name: "frostdb_lsm_level_size_bytes",
+			Help: "The size of the level in bytes.",
+		}, []string{"level"}),
+
+		CompactionDuration: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
+			Name:                        "frostdb_lsm_compaction_total_duration_seconds",
+			Help:                        "Total compaction duration",
+			NativeHistogramBucketFactor: 1.1,
+		}),
+	}
+}
+
+// NewLSM returns an LSM-like index of len(levels) levels.
+func NewLSM(prefix string, levels []*LevelConfig, options ...LSMOption) (*LSM, error) {
+	if err := validateLevels(levels); err != nil {
+		return nil, err
+	}
+
+	lsm := &LSM{
+		prefix:     prefix,
+		levels:     NewList(L0),
+		sizes:      make([]atomic.Int64, len(levels)),
+		configs:    levels,
+		compacting: &atomic.Bool{},
+		logger:     log.NewNopLogger(),
+	}
+
+	for _, opt := range options {
+		opt(lsm)
+	}
+
+	// Reverse iterate (due to prepend) to create the chain of sentinel nodes.
+	for i := len(levels) - 1; i > 0; i-- {
+		lsm.levels.Sentinel(levels[i].Level)
+	}
+
+	if lsm.metrics == nil {
+		lsm.metrics = NewLSMMetrics(prometheus.NewRegistry())
+	}
+
+	return lsm, nil
+}
+
+// Size returns the total size of the index in bytes.
+func (l *LSM) Size() int64 {
+	var size int64
+	for i := range l.sizes {
+		size += l.sizes[i].Load()
+	}
+	return size
+}
+
+func validateLevels(levels []*LevelConfig) error {
+	for i, l := range levels {
+		if int(l.Level) != i {
+			return fmt.Errorf("level %d is not in order", l.Level)
+		}
+
+		switch i {
+		case len(levels) - 1:
+			if l.Compact != nil {
+				return fmt.Errorf("level %d is the last level and should not have a compact function", l.Level)
+			}
+		default:
+			if l.Compact == nil {
+				return fmt.Errorf("level %d is not the last level and should have a compact function", l.Level)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (l *LSM) MaxLevel() SentinelType {
+	return SentinelType(len(l.configs) - 1)
+}
+
+func (l *LSM) Add(tx uint64, record arrow.Record) {
+	record.Retain()
+	size := util.TotalRecordSize(record)
+	l.levels.Prepend(parts.NewArrowPart(tx, record, int(size), nil))
+	l0 := l.sizes[L0].Add(int64(size))
+	l.metrics.LevelSize.WithLabelValues(L0.String()).Set(float64(l0))
+	if l0 >= l.configs[L0].MaxSize {
+		if l.compacting.CompareAndSwap(false, true) {
+			go func() {
+				_ = l.compact()
+			}()
+		}
+	}
+}
+
+// InsertPart inserts a part into the LSM tree. It will be inserted into the correct level. It does not check if the insert should cause a compaction.
+// This should only be used during snapshot recovery.
+func (l *LSM) InsertPart(level SentinelType, part *parts.Part) {
+	l.findLevel(level).Prepend(part)
+	size := l.sizes[level].Add(int64(part.Size()))
+	l.metrics.LevelSize.WithLabelValues(level.String()).Set(float64(size))
+}
+
+func (l *LSM) String() string {
+	s := ""
+	for i := range l.sizes {
+		s += fmt.Sprintf("L%v: %d ", i, l.sizes[i].Load())
+	}
+	s += "\n"
+	s += l.levels.String()
+	return s
+}
+
+func (l *LSM) Prefixes(ctx context.Context, prefix string) ([]string, error) {
+	return []string{l.prefix}, nil
+}
+
+func (l *LSM) Iterate(iter func(node *Node) bool) {
+	l.levels.Iterate(iter)
+}
+
+func (l *LSM) Scan(ctx context.Context, _ string, _ *dynparquet.Schema, filter logicalplan.Expr, tx uint64, callback func(context.Context, any) error) error {
+	l.RLock()
+	defer l.RUnlock()
+
+	booleanFilter, err := expr.BooleanExpr(filter)
+	if err != nil {
+		return fmt.Errorf("boolean expr: %w", err)
+	}
+	var iterError error
+	l.levels.Iterate(func(node *Node) bool {
+		if node.part == nil { // encountered a sentinel node; continue on
+			return true
+		}
+
+		if node.part.TX() > tx { // skip parts that are newer than this transaction
+			return true
+		}
+
+		if r := node.part.Record(); r != nil {
+			r.Retain()
+			if err := callback(ctx, r); err != nil {
+				iterError = err
+				return false
+			}
+			return true
+		}
+
+		buf, err := node.part.AsSerializedBuffer(nil)
+		if err != nil {
+			iterError = err
+			return false
+		}
+
+		for i := 0; i < buf.NumRowGroups(); i++ {
+			rg := buf.DynamicRowGroup(i)
+			mayContainUsefulData, err := booleanFilter.Eval(rg)
+			if err != nil {
+				iterError = err
+				return false
+			}
+
+			if mayContainUsefulData {
+				if err := callback(ctx, rg); err != nil {
+					iterError = err
+					return false
+				}
+			}
+		}
+		return true
+	})
+	return iterError
+}
+
+// TODO: this should be changed to just retain the sentinel nodes in the lsm struct to do an O(1) lookup.
+func (l *LSM) findLevel(level SentinelType) *Node {
+	var list *Node
+	l.levels.Iterate(func(node *Node) bool {
+		if node.part == nil && node.sentinel == level {
+			list = node
+			return false
+		}
+		return true
+	})
+
+	return list
+}
+
+// findNode returns the node that points to node.
+func (l *LSM) findNode(node *Node) *Node {
+	var list *Node
+	l.levels.Iterate(func(n *Node) bool {
+		if n.next.Load() == node {
+			list = n
+			return false
+		}
+		return true
+	})
+
+	return list
+}
+
+func (l *LSM) Compact() error {
+	for !l.compacting.CompareAndSwap(false, true) { // TODO: should backoff retry this probably
+	}
+	return l.compact()
+}
+
+func (l *LSM) Rotate(level SentinelType, externalWriter func([]*parts.Part) (*parts.Part, int64, int64, error)) error {
+	for !l.compacting.CompareAndSwap(false, true) { // TODO: should backoff retry this probably
+	}
+	defer l.compacting.Store(false)
+	start := time.Now()
+	defer func() {
+		l.metrics.CompactionDuration.Observe(time.Since(start).Seconds())
+	}()
+
+	for i := 0; i < len(l.configs)-1; i++ {
+		if err := l.merge(SentinelType(i), nil); err != nil {
+			return err
+		}
+	}
+	return l.merge(level, externalWriter)
+}
+
+// Merge will merge the given level into an arrow record for the next level using the configured Compact function for the given level.
+// If this is the max level of the LSM an external writer must be provided to write the merged part elsewhere.
+func (l *LSM) merge(level SentinelType, externalWriter func([]*parts.Part) (*parts.Part, int64, int64, error)) error {
+	if int(level) > len(l.configs) {
+		return fmt.Errorf("level %d does not exist", level)
+	}
+	if int(level) == len(l.configs)-1 && externalWriter == nil {
+		return fmt.Errorf("cannot merge the last level without an external writer")
+	}
+	l.metrics.Compactions.WithLabelValues(level.String()).Inc()
+
+	nodeList := []*Node{}
+	var next *Node
+	compact := l.findLevel(level)
+	var iterErr error
+	compact.Iterate(func(node *Node) bool {
+		if node.part == nil { // sentinel encountered
+			switch {
+			case node.sentinel == level: // the sentinel for the beginning of the list
+				return true
+			case node.sentinel == level+1:
+				next = node.next.Load() // skip the sentinel to combine the lists
+			default:
+				next = node
+			}
+			return false
+		}
+
+		nodeList = append(nodeList, node)
+		return true
+	})
+	if iterErr != nil {
+		return iterErr
+	}
+
+	if len(nodeList) == 0 {
+		return nil
+	}
+
+	var size int64
+	var compactedSize int64
+	var compacted []*parts.Part
+	var err error
+	mergeList := make([]*parts.Part, 0, len(nodeList))
+	for _, node := range nodeList {
+		mergeList = append(mergeList, node.part)
+	}
+	s := &Node{
+		sentinel: level + 1,
+	}
+	if externalWriter != nil {
+		_, size, _, err = externalWriter(mergeList)
+		if err != nil {
+			return err
+		}
+		// Drop compacted files from list
+		if next != nil {
+			s.next.Store(next)
+		}
+	} else {
+		compacted, size, compactedSize, err = l.configs[level].Compact(mergeList)
+		if err != nil {
+			return err
+		}
+
+		// Create new list for the compacted parts.
+		compactedList := &Node{
+			part: compacted[0],
+		}
+		node := compactedList
+		for _, p := range compacted[1:] {
+			node.next.Store(&Node{
+				part: p,
+			})
+			node = node.next.Load()
+		}
+		s.next.Store(compactedList)
+		if next != nil {
+			node.next.Store(next)
+		}
+		l.sizes[level+1].Add(int64(compactedSize))
+		l.metrics.LevelSize.WithLabelValues(SentinelType(level + 1).String()).Set(float64(l.sizes[level+1].Load()))
+	}
+
+	// Replace the compacted list with the new list
+	// find the node that points to the first node in our compacted list.
+	node := l.findNode(nodeList[0])
+	for !node.next.CompareAndSwap(nodeList[0], s) {
+		// This can happen at most once in the scenario where a new part is added to the L0 list while we are trying to replace it.
+		node = l.findNode(nodeList[0])
+	}
+	l.sizes[level].Add(-int64(size))
+	l.metrics.LevelSize.WithLabelValues(level.String()).Set(float64(l.sizes[level].Load()))
+
+	// release the old parts
+	l.Lock()
+	defer l.Unlock()
+	for _, part := range mergeList {
+		if r := part.Record(); r != nil {
+			r.Release()
+		}
+	}
+
+	return nil
+}
+
+// compact is a cascading compaction routine. It will start at the lowest level and compact until the next level is either the max level or the next level does not exceed the max size.
+// compact can not be run concurrently.
+func (l *LSM) compact() error {
+	defer l.compacting.Store(false)
+	start := time.Now()
+	defer func() {
+		l.metrics.CompactionDuration.Observe(time.Since(start).Seconds())
+	}()
+
+	for i := 0; i < len(l.configs)-1; i++ {
+		if l.sizes[i].Load() >= l.configs[i].MaxSize {
+			if err := l.merge(SentinelType(i), nil); err != nil {
+				level.Error(l.logger).Log("msg", "failed to merge level", "level", i, "err", err)
+				return err
+			}
+		}
+	}
+
+	return nil
+}
diff --git a/index/lsm_list.go b/index/lsm_list.go
new file mode 100644
index 0000000..c54511c
--- /dev/null
+++ b/index/lsm_list.go
@@ -0,0 +1,104 @@
+package index
+
+import (
+	"fmt"
+	"sync/atomic"
+
+	"github.com/polarsignals/frostdb/parts"
+)
+
+type SentinelType int
+
+const (
+	L0 SentinelType = iota
+	L1
+	L2
+)
+
+func (s SentinelType) String() string {
+	return fmt.Sprintf("L%v", int(s))
+}
+
+// Node is a Part that is a part of a linked-list.
+type Node struct {
+	next atomic.Pointer[Node]
+	part *parts.Part
+
+	sentinel SentinelType // sentinel nodes contain no parts, and are to indicate the start of a new sub list
+}
+
+func (n *Node) Part() *parts.Part {
+	return n.part
+}
+
+func (n *Node) String() string {
+	if n.part == nil {
+		if n.next.Load() == nil {
+			return fmt.Sprintf("[%v]", n.sentinel)
+		}
+		return fmt.Sprintf("[%v]->%v", n.sentinel, n.next.Load().String())
+	}
+
+	if n.part.Record() != nil {
+		if n.next.Load() == nil {
+			return fmt.Sprintf("[%v]", n.part.Record().NumRows())
+		}
+		return fmt.Sprintf("[%v]->%v", n.part.Record().NumRows(), n.next.Load().String())
+	}
+
+	b, _ := n.part.AsSerializedBuffer(nil)
+	if n.next.Load() == nil {
+		return fmt.Sprintf("[%v]", b.NumRows())
+	}
+	return fmt.Sprintf("[%v]->%v", b.NumRows(), n.next.Load().String())
+}
+
+// NewList creates a new part list using atomic constructs.
+func NewList(sentinel SentinelType) *Node {
+	p := &Node{
+		sentinel: sentinel,
+	}
+	return p
+}
+
+// Sentinel adds a new sentinel node to the list, and returns the sub list starting from that sentinel.
+func (n *Node) Sentinel(s SentinelType) *Node {
+	return n.prepend(&Node{
+		sentinel: s,
+	})
+}
+
+// Prepend a node onto the front of the list.
+func (n *Node) Prepend(part *parts.Part) *Node {
+	return n.prepend(&Node{
+		part: part,
+	})
+}
+
+func (n *Node) prepend(node *Node) *Node {
+	for { // continue until a successful compare and swap occurs
+		next := n.next.Load()
+		node.next.Store(next)
+		if n.next.CompareAndSwap(next, node) {
+			return node
+		}
+	}
+}
+
+// Iterate accesses every node in the list.
+func (n *Node) Iterate(iterate func(*Node) bool) {
+	if !iterate(n) {
+		return
+	}
+
+	node := n.next.Load()
+	for {
+		if node == nil {
+			return
+		}
+		if !iterate(node) {
+			return
+		}
+		node = node.next.Load()
+	}
+}
diff --git a/binaryscalarexpr.go b/query/expr/binaryscalarexpr.go
similarity index 99%
rename from binaryscalarexpr.go
rename to query/expr/binaryscalarexpr.go
index 0569b1b..5d53764 100644
--- a/binaryscalarexpr.go
+++ b/query/expr/binaryscalarexpr.go
@@ -1,4 +1,4 @@
-package frostdb
+package expr
 
 import (
 	"errors"
diff --git a/filter.go b/query/expr/filter.go
similarity index 99%
rename from filter.go
rename to query/expr/filter.go
index 2839ab9..8b8b542 100644
--- a/filter.go
+++ b/query/expr/filter.go
@@ -1,4 +1,4 @@
-package frostdb
+package expr
 
 import (
 	"errors"
diff --git a/snapshot.go b/snapshot.go
index 2780164..7a4e559 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -13,20 +13,20 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/oklog/ulid"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/arrow/go/v14/arrow/ipc"
 	"github.com/apache/arrow/go/v14/arrow/util"
 	"github.com/go-kit/log/level"
-	"github.com/google/btree"
-	"github.com/oklog/ulid"
-	"google.golang.org/protobuf/proto"
 
 	"github.com/polarsignals/frostdb/dynparquet"
 	snapshotpb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/snapshot/v1alpha1"
 	tablepb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/table/v1alpha1"
 	walpb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/wal/v1alpha1"
+	"github.com/polarsignals/frostdb/index"
 	"github.com/polarsignals/frostdb/parts"
 )
 
@@ -425,54 +425,53 @@ func WriteSnapshot(ctx context.Context, tx uint64, txnMetadata []byte, db *DB, w
 			}
 
 			var ascendErr error
-			block.Index().Ascend(func(i btree.Item) bool {
+			block.Index().Iterate(func(node *index.Node) bool {
 				granuleMeta := &snapshotpb.Granule{}
-				i.(*Granule).PartsForTx(tx, func(p *parts.Part) bool {
-					partMeta := &snapshotpb.Part{
-						StartOffset:     int64(offW.offset),
-						Tx:              p.TX(),
-						CompactionLevel: uint64(p.CompactionLevel()),
+				if node.Part() == nil {
+					return true
+				}
+				p := node.Part()
+				partMeta := &snapshotpb.Part{
+					StartOffset:     int64(offW.offset),
+					Tx:              p.TX(),
+					CompactionLevel: uint64(p.CompactionLevel()),
+				}
+				if err := func() error {
+					if err := ctx.Err(); err != nil {
+						return err
 					}
-					if err := func() error {
-						if err := ctx.Err(); err != nil {
-							return err
-						}
-						schema := t.schema
-
-						if record := p.Record(); record != nil {
-							partMeta.Encoding = snapshotpb.Part_ENCODING_ARROW
-							recordWriter := ipc.NewWriter(
-								w,
-								ipc.WithSchema(record.Schema()),
-							)
-							defer recordWriter.Close()
-							return recordWriter.Write(record)
-						}
-						partMeta.Encoding = snapshotpb.Part_ENCODING_PARQUET
+					schema := t.schema
+
+					if record := p.Record(); record != nil {
+						partMeta.Encoding = snapshotpb.Part_ENCODING_ARROW
+						recordWriter := ipc.NewWriter(
+							w,
+							ipc.WithSchema(record.Schema()),
+						)
+						defer recordWriter.Close()
+						return recordWriter.Write(record)
+					}
+					partMeta.Encoding = snapshotpb.Part_ENCODING_PARQUET
 
-						buf, err := p.AsSerializedBuffer(schema)
-						if err != nil {
-							return err
-						}
+					buf, err := p.AsSerializedBuffer(schema)
+					if err != nil {
+						return err
+					}
 
-						f := buf.ParquetFile()
-						if _, err := io.Copy(
-							w, io.NewSectionReader(f, 0, f.Size()),
-						); err != nil {
-							return err
-						}
-						return nil
-					}(); err != nil {
-						ascendErr = err
-						return false
+					f := buf.ParquetFile()
+					if _, err := io.Copy(
+						w, io.NewSectionReader(f, 0, f.Size()),
+					); err != nil {
+						return err
 					}
-					partMeta.EndOffset = int64(offW.offset)
-					granuleMeta.PartMetadata = append(granuleMeta.PartMetadata, partMeta)
-					return true
-				})
-				if len(granuleMeta.PartMetadata) > 0 {
-					tableMeta.GranuleMetadata = append(tableMeta.GranuleMetadata, granuleMeta)
+					return nil
+				}(); err != nil {
+					ascendErr = err
+					return false
 				}
+				partMeta.EndOffset = int64(offW.offset)
+				granuleMeta.PartMetadata = append(granuleMeta.PartMetadata, partMeta)
+				tableMeta.GranuleMetadata = append(tableMeta.GranuleMetadata, granuleMeta) // TODO: we have one part per granule now
 				return true
 			})
 			metadata.TableMetadata = append(metadata.TableMetadata, tableMeta)
@@ -564,9 +563,6 @@ func loadSnapshot(ctx context.Context, db *DB, r io.ReaderAt, size int64) ([]byt
 		return nil, err
 	}
 
-	db.compactorPool.pause()
-	defer db.compactorPool.resume()
-
 	for i, tableMeta := range footer.TableMetadata {
 		if err := func() error {
 			var schemaMsg proto.Message
@@ -604,13 +600,12 @@ func loadSnapshot(ctx context.Context, db *DB, r io.ReaderAt, size int64) ([]byt
 			block := table.active
 			block.mtx.Lock()
 			block.ulid = blockUlid
-			block.size.Store(tableMeta.ActiveBlock.Size)
 			// Store the last snapshot size so a snapshot is not triggered right
 			// after loading this snapshot.
 			block.lastSnapshotSize.Store(tableMeta.ActiveBlock.Size)
 			block.minTx = tableMeta.ActiveBlock.MinTx
 			block.prevTx = tableMeta.ActiveBlock.PrevTx
-			newIdx := block.Index().Clone()
+			newIdx := block.Index()
 			block.mtx.Unlock()
 			table.mtx.Unlock()
 
@@ -651,16 +646,13 @@ func loadSnapshot(ctx context.Context, db *DB, r io.ReaderAt, size int64) ([]byt
 					}
 				}
 
-				granule, err := NewGranule(table.schema, resultParts...)
-				if err != nil {
-					return err
+				for _, part := range resultParts {
+					if part.Record() != nil { // TODO: eventually store the level in the part metadata
+						newIdx.InsertPart(index.L0, part)
+					} else {
+						newIdx.InsertPart(index.L1, part)
+					}
 				}
-				newIdx.ReplaceOrInsert(granule)
-			}
-
-			// This shouldn't be necessary since compactions were paused and no
-			// inserts should be happening, but err on the side of caution.
-			for !block.index.CompareAndSwap(block.Index(), newIdx) {
 			}
 
 			return nil
diff --git a/store.go b/store.go
index bd6ede1..b09daa5 100644
--- a/store.go
+++ b/store.go
@@ -16,6 +16,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/polarsignals/frostdb/dynparquet"
+	"github.com/polarsignals/frostdb/query/expr"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 	"github.com/polarsignals/frostdb/storage"
 )
@@ -139,7 +140,7 @@ func (b *DefaultObjstoreBucket) Scan(ctx context.Context, prefix string, _ *dynp
 	span.SetAttributes(attribute.Int64("lastBlockTimestamp", int64(lastBlockTimestamp)))
 	defer span.End()
 
-	f, err := BooleanExpr(filter)
+	f, err := expr.BooleanExpr(filter)
 	if err != nil {
 		return err
 	}
@@ -183,7 +184,7 @@ func (b *DefaultObjstoreBucket) openBlockFile(ctx context.Context, blockName str
 }
 
 // ProcessFile will process a bucket block parquet file.
-func (b *DefaultObjstoreBucket) ProcessFile(ctx context.Context, blockDir string, lastBlockTimestamp uint64, filter TrueNegativeFilter, callback func(context.Context, any) error) error {
+func (b *DefaultObjstoreBucket) ProcessFile(ctx context.Context, blockDir string, lastBlockTimestamp uint64, filter expr.TrueNegativeFilter, callback func(context.Context, any) error) error {
 	ctx, span := b.tracer.Start(ctx, "Source/IterateBucketBlocks/Iter/ProcessFile")
 	defer span.End()
 
@@ -225,7 +226,7 @@ func (b *DefaultObjstoreBucket) ProcessFile(ctx context.Context, blockDir string
 	return b.filterRowGroups(ctx, buf, filter, callback)
 }
 
-func (b *DefaultObjstoreBucket) filterRowGroups(ctx context.Context, buf *dynparquet.SerializedBuffer, filter TrueNegativeFilter, callback func(context.Context, any) error) error {
+func (b *DefaultObjstoreBucket) filterRowGroups(ctx context.Context, buf *dynparquet.SerializedBuffer, filter expr.TrueNegativeFilter, callback func(context.Context, any) error) error {
 	_, span := b.tracer.Start(ctx, "Source/filterRowGroups")
 	defer span.End()
 	span.SetAttributes(attribute.Int("row_groups", buf.NumRowGroups()))
diff --git a/table.go b/table.go
index a6baa18..1bbc866 100644
--- a/table.go
+++ b/table.go
@@ -1,11 +1,11 @@
 package frostdb
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"math/rand"
 	"path/filepath"
 	"runtime"
@@ -17,11 +17,9 @@ import (
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
 	"github.com/apache/arrow/go/v14/arrow/memory"
-	"github.com/apache/arrow/go/v14/arrow/util"
 	"github.com/dustin/go-humanize"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
-	"github.com/google/btree"
 	"github.com/oklog/ulid"
 	"github.com/parquet-go/parquet-go"
 	"github.com/prometheus/client_golang/prometheus"
@@ -31,12 +29,12 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/polarsignals/frostdb/bufutils"
 	"github.com/polarsignals/frostdb/dynparquet"
 	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
 	schemav2pb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha2"
 	tablepb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/table/v1alpha1"
 	walpb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/wal/v1alpha1"
+	"github.com/polarsignals/frostdb/index"
 	"github.com/polarsignals/frostdb/parts"
 	"github.com/polarsignals/frostdb/pqarrow"
 	"github.com/polarsignals/frostdb/query/logicalplan"
@@ -46,8 +44,13 @@ import (
 )
 
 var (
-	ErrNoSchema     = fmt.Errorf("no schema")
-	ErrTableClosing = fmt.Errorf("table closing")
+	ErrNoSchema        = fmt.Errorf("no schema")
+	ErrTableClosing    = fmt.Errorf("table closing")
+	DefaultIndexConfig = []*index.LevelConfig{
+		{Level: index.L0, MaxSize: 1024 * 1024 * 15},  // Compact to Parquet after 15MiB of data
+		{Level: index.L1, MaxSize: 1024 * 1024 * 128}, // Compact to a single Parquet after 128MiB of Parquet files
+		{Level: index.L2, MaxSize: 1024 * 1024 * 512}, // Final level. Rotate after 512MiB of Parquet files
+	}
 )
 
 type ErrWriteRow struct{ err error }
@@ -187,11 +190,10 @@ type TableBlock struct {
 	minTx  uint64
 	prevTx uint64
 
-	size *atomic.Int64
 	// lastSnapshotSize keeps track of the size of the block when it last
 	// triggered a snapshot.
 	lastSnapshotSize atomic.Int64
-	index            *atomic.Pointer[btree.BTree] // *btree.BTree
+	index            *index.LSM
 
 	pendingWritersWg sync.WaitGroup
 	pendingReadersWg sync.WaitGroup
@@ -200,19 +202,15 @@ type TableBlock struct {
 }
 
 type tableMetrics struct {
-	blockPersisted            prometheus.Counter
-	blockRotated              prometheus.Counter
-	granulesCreated           prometheus.Counter
-	compactions               prometheus.Counter
-	granulesSplits            prometheus.Counter
-	rowsInserted              prometheus.Counter
-	zeroRowsInserted          prometheus.Counter
-	granulesCompactionAborted prometheus.Counter
-	rowInsertSize             prometheus.Histogram
-	lastCompletedBlockTx      prometheus.Gauge
-	numParts                  prometheus.Gauge
-	unsortedInserts           prometheus.Counter
-	compactionMetrics         *compactionMetrics
+	blockPersisted       prometheus.Counter
+	blockRotated         prometheus.Counter
+	rowsInserted         prometheus.Counter
+	zeroRowsInserted     prometheus.Counter
+	rowInsertSize        prometheus.Histogram
+	lastCompletedBlockTx prometheus.Gauge
+	numParts             prometheus.Gauge
+
+	indexMetrics *index.LSMMetrics
 }
 
 func schemaFromTableConfig(tableConfig *tablepb.TableConfig) (*dynparquet.Schema, error) {
@@ -279,22 +277,6 @@ func newTable(
 				Name: "frostdb_table_blocks_rotated_total",
 				Help: "Number of table blocks that have been rotated.",
 			}),
-			granulesCreated: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-				Name: "frostdb_table_granules_created_total",
-				Help: "Number of granules created.",
-			}),
-			compactions: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-				Name: "frostdb_table_granules_compactions_total",
-				Help: "Number of granule compactions.",
-			}),
-			granulesSplits: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-				Name: "frostdb_table_granules_splits_total",
-				Help: "Number of granules splits executed.",
-			}),
-			granulesCompactionAborted: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-				Name: "frostdb_table_granules_compaction_aborted_total",
-				Help: "Number of aborted granules compaction.",
-			}),
 			rowsInserted: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 				Name: "frostdb_table_rows_inserted_total",
 				Help: "Number of rows inserted into table.",
@@ -312,11 +294,7 @@ func newTable(
 				Name: "frostdb_table_last_completed_block_tx",
 				Help: "Last completed block transaction.",
 			}),
-			unsortedInserts: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-				Name: "frostdb_table_unsorted_inserts_total",
-				Help: "The number of times a buffer to insert was not in sorted order.",
-			}),
-			compactionMetrics: newCompactionMetrics(reg, float64(db.columnStore.granuleSizeBytes)),
+			indexMetrics: index.NewLSMMetrics(reg),
 		},
 	}
 
@@ -327,16 +305,6 @@ func newTable(
 
 	t.pendingBlocks = make(map[*TableBlock]struct{})
 
-	promauto.With(reg).NewGaugeFunc(prometheus.GaugeOpts{
-		Name: "frostdb_table_index_size",
-		Help: "Number of granules in the table index currently.",
-	}, func() float64 {
-		if active := t.ActiveBlock(); active != nil {
-			return float64(active.Index().Len())
-		}
-		return 0
-	})
-
 	promauto.With(reg).NewGaugeFunc(prometheus.GaugeOpts{
 		Name: "frostdb_table_active_block_size",
 		Help: "Size of the active table block in bytes.",
@@ -388,16 +356,7 @@ func (t *Table) dropPendingBlock(block *TableBlock) {
 	block.pendingReadersWg.Wait()
 	block.pendingWritersWg.Wait()
 
-	block.Index().Ascend(func(i btree.Item) bool {
-		g := i.(*Granule)
-		g.PartsForTx(math.MaxUint64, func(p *parts.Part) bool {
-			if r := p.Record(); r != nil {
-				r.Release()
-			}
-			return true
-		})
-		return true
-	})
+	// TODO thor call Release on all records in the block...
 }
 
 func (t *Table) writeBlock(block *TableBlock, skipPersist, snapshotDB bool) {
@@ -715,10 +674,7 @@ func (t *Table) Iterator(
 						if r.NumRows() == 0 {
 							return nil
 						}
-						if err := callback(ctx, r); err != nil {
-							return err
-						}
-						return nil
+						return callback(ctx, r)
 					}
 
 					switch t := rg.(type) {
@@ -870,25 +826,32 @@ func generateULID() ulid.ULID {
 }
 
 func newTableBlock(table *Table, prevTx, tx uint64, id ulid.ULID) (*TableBlock, error) {
-	index := atomic.Pointer[btree.BTree]{}
-	index.Store(btree.New(table.db.columnStore.indexDegree))
-
-	return &TableBlock{
+	tb := &TableBlock{
 		table:  table,
-		index:  &index,
 		mtx:    &sync.RWMutex{},
 		ulid:   id,
-		size:   &atomic.Int64{},
 		logger: table.logger,
 		tracer: table.tracer,
 		minTx:  tx,
 		prevTx: prevTx,
-	}, nil
+	}
+
+	var err error
+	tb.index, err = index.NewLSM(
+		table.name,
+		table.configureLSMLevels(table.db.columnStore.indexConfig),
+		index.LSMWithMetrics(table.metrics.indexMetrics),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return tb, nil
 }
 
 // EnsureCompaction forces a TableBlock compaction.
 func (t *TableBlock) EnsureCompaction() error {
-	return t.compact(t.table.db.columnStore.compactionConfig)
+	return t.index.Compact()
 }
 
 func (t *TableBlock) InsertRecord(ctx context.Context, tx uint64, record arrow.Record) error {
@@ -905,318 +868,36 @@ func (t *TableBlock) InsertRecord(ctx context.Context, tx uint64, record arrow.R
 	record = dynparquet.PrehashColumns(t.table.schema, record)
 	defer record.Release()
 
-	if err := t.insertRecordToGranules(tx, record); err != nil {
-		return fmt.Errorf("failed to insert record into granules: %w", err)
-	}
-
+	t.index.Add(tx, record)
+	t.table.metrics.numParts.Inc()
 	return nil
 }
 
-// updateIndex updates that TableBlock's index in a thread safe manner to the result of calling the update fn with the current index.
-func (t *TableBlock) updateIndex(update func(index *btree.BTree) error) error {
-	var replaceErr error
-	for ok := false; !ok; {
-		ok = func() bool {
-			old := t.Index()
-			t.mtx.Lock()
-			newIndex := old.Clone()
-			t.mtx.Unlock()
-
-			if err := update(newIndex); err != nil {
-				replaceErr = err
-				return true
-			}
-
-			return t.index.CompareAndSwap(old, newIndex)
-		}()
-	}
-	return replaceErr
-}
-
-// RowGroupIterator iterates in order over all granules in the table.
-// It stops iterating when the iterator function returns false.
-func (t *TableBlock) RowGroupIterator(
-	ctx context.Context,
-	tx uint64,
-	filter TrueNegativeFilter,
-	rowGroups chan<- any,
-) error {
-	ctx, span := t.tracer.Start(ctx, "TableBlock/RowGroupIterator")
-	span.SetAttributes(attribute.Int64("tx", int64(tx))) // Attributes don't support uint64...
-	defer span.End()
-
-	index := t.Index()
-
-	var err error
-	index.Ascend(func(i btree.Item) bool {
-		g := i.(*Granule)
-		g.Collect(ctx, tx, filter, rowGroups)
-		return true
-	})
-
-	return err
-}
-
 // Size returns the cumulative size of all buffers in the table. This is roughly the size of the table in bytes.
 func (t *TableBlock) Size() int64 {
-	return t.size.Load()
+	return t.index.Size()
 }
 
 // Index provides atomic access to the table index.
-func (t *TableBlock) Index() *btree.BTree {
-	return (*btree.BTree)(t.index.Load())
-}
-
-func (t *TableBlock) insertRecordToGranules(tx uint64, record arrow.Record) error {
-	ps := t.table.schema
-	numRows := record.NumRows()
-	idx := numRows - 1
-	dynCols := pqarrow.RecordDynamicCols(record)
-	pooledSchema, err := ps.GetDynamicParquetSchema(dynCols)
-	if err != nil {
-		return err
-	}
-	defer ps.PutPooledParquetSchema(pooledSchema)
-	row, err := pqarrow.RecordToDynamicRow(ps, pooledSchema.Schema, record, dynCols, int(idx))
-	if err != nil {
-		if err == io.EOF {
-			level.Debug(t.logger).Log("msg", "inserted record with no rows")
-			return nil
-		}
-		return err
-	}
-
-	// recordSizePerRow is the rough estimate of the size of each row in the record.
-	recordSizePerRow := int(util.TotalRecordSize(record) / int64(numRows))
-
-	var ascendErr error
-	t.Index().DescendLessOrEqual(
-		btreeComparableDynamicRow{
-			schema:     ps,
-			DynamicRow: row,
-		},
-		func(i btree.Item) bool {
-			g := i.(*Granule)
-			// Descend rows to insert until we find a row that does not belong in this granule.
-			n := 0
-			for ; idx >= 0; idx-- {
-				row, err := pqarrow.RecordToDynamicRow(ps, pooledSchema.Schema, record, dynCols, int(idx))
-				if err != nil {
-					ascendErr = err
-					return false
-				}
-				if ps.RowLessThan(row, g.Least()) {
-					if n > 0 {
-						if _, err := g.Append(parts.NewArrowPart(tx, record.NewSlice(idx+1, idx+1+int64(n)), recordSizePerRow*n, ps)); err != nil {
-							ascendErr = err
-							return false
-						}
-						t.table.metrics.numParts.Add(float64(1))
-					}
-					// Go on to the next granule.
-					return true
-				}
-				n++
-			}
-			// If we got here, all rows were exhaused in the loop above.
-			if n > 0 {
-				if _, err := g.Append(parts.NewArrowPart(tx, record.NewSlice(idx+1, idx+1+int64(n)), recordSizePerRow*n, ps)); err != nil {
-					ascendErr = err
-					return false
-				}
-				t.table.metrics.numParts.Add(float64(1))
-			}
-			return false
-		},
-	)
-	if ascendErr != nil {
-		return err
-	}
-
-	if idx < 0 {
-		// All rows exhausted.
-		t.size.Add(util.TotalRecordSize(record))
-		return nil
-	}
-
-	g, err := NewGranule(ps, parts.NewArrowPart(tx, record.NewSlice(0, idx+1), recordSizePerRow*int(idx+1), ps))
-	if err != nil {
-		return fmt.Errorf("new granule failed: %w", err)
-	}
-
-	if err := t.updateIndex(func(index *btree.BTree) error {
-		index.ReplaceOrInsert(g)
-		return nil
-	}); err != nil {
-		return err
-	}
-	t.table.metrics.granulesCreated.Inc()
-	t.table.metrics.numParts.Add(float64(1))
-	t.size.Add(util.TotalRecordSize(record))
-	return nil
-}
-
-type btreeComparableDynamicRow struct {
-	schema *dynparquet.Schema
-	*dynparquet.DynamicRow
-}
-
-func (r btreeComparableDynamicRow) Less(than btree.Item) bool {
-	return r.schema.RowLessThan(r.DynamicRow, than.(*Granule).Least())
-}
-
-// addPartToGranule finds the corresponding granule it belongs to in a sorted list of Granules.
-func addPartToGranule(granules []*Granule, p *parts.Part) error {
-	row, err := p.Least()
-	if err != nil {
-		return err
-	}
-
-	var prev *Granule
-	for _, g := range granules {
-		if g.schema.RowLessThan(row, g.Least()) {
-			if prev != nil {
-				if _, err := prev.Append(p); err != nil {
-					return err
-				}
-				return nil
-			}
-		} else {
-			prev = g
-		}
-	}
-
-	if prev != nil {
-		// Save part to prev
-		if _, err := prev.Append(p); err != nil {
-			return err
-		}
-	} else {
-		// NOTE: this should never happen
-		panic("programming error; unable to find granule for part")
-	}
-
-	return nil
-}
-
-// abortCompaction resets state set on compaction so that a granule may be
-// compacted again.
-func (t *TableBlock) abortCompaction(granule *Granule) {
-	t.table.metrics.granulesCompactionAborted.Inc()
-	for {
-		// Unmark pruned, so that we can compact the granule in the future.
-		if granule.metadata.pruned.CompareAndSwap(1, 0) {
-			return
-		}
-	}
+func (t *TableBlock) Index() *index.LSM {
+	return t.index
 }
 
 // Serialize the table block into a single Parquet file.
-// The Serialize function will walk all Granules in the block, compact each Granule into a sorted set of Row groups, then write those
-// row groups to the final Parquet file, repeating for each Granule. This leverages the fact that the index has alreayd sorted the Granules
-// so no need to sort them further than the intra-granule parts.
 func (t *TableBlock) Serialize(writer io.Writer) error {
-	var ascendErr error
-	dynCols := map[string][]string{} // NOTE: it would be nice if we didn't have to go back and find all the dynamic columns...
-	index := t.Index()
-	index.Ascend(func(i btree.Item) bool {
-		g := i.(*Granule)
-
-		g.PartsForTx(math.MaxUint64, func(p *parts.Part) bool {
-			buf, err := p.AsSerializedBuffer(t.table.schema)
-			if err != nil {
-				ascendErr = err
-				return false
-			}
-
-			for key, vals := range buf.DynamicColumns() {
-				dynCols[key] = append(dynCols[key], vals...)
-			}
-			return true
-		})
-
-		return true
-	})
-	if ascendErr != nil {
-		return ascendErr
-	}
-
-	dynCols = bufutils.Dedupe(dynCols)
-	rowWriter, err := t.rowWriter(writer, dynCols)
-	if err != nil {
-		return fmt.Errorf("create row writer: %w", err)
-	}
-	defer rowWriter.close()
-
-	cols := t.table.schema.ParquetSortingColumns(dynCols)
-	pooledSchema, err := t.table.schema.GetDynamicParquetSchema(dynCols)
-	if err != nil {
-		return err
-	}
-	defer t.table.schema.PutPooledParquetSchema(pooledSchema)
-	ps := pooledSchema.Schema
-
-	// Merge all parts in a Granule and write that granule to the file
-	index.Ascend(func(i btree.Item) bool {
-		g := i.(*Granule)
-
-		// collect parts based on compaction level; we do this because level1 parts are already non-overlapping.
-		lvl0, lvl1, _, stats, err := collectPartsForCompaction(math.MaxUint64, g.parts)
-		if err != nil {
-			ascendErr = err
-			return false
-		}
-
-		// Compact the level 0 parts
-		if len(lvl0) > 0 {
-			lvl1, err = compactLevel0IntoLevel1(t, math.MaxUint64, lvl0, lvl1, 1.0, &stats)
-			if err != nil {
-				ascendErr = err
-				return false
-			}
-		}
-
-		sorter := parts.NewPartSorter(t.table.schema, lvl1)
-		sort.Sort(sorter)
-		if err := sorter.Err(); err != nil {
-			ascendErr = err
-			return false
-		}
-
-		// Write the sorted parts
-		for _, p := range lvl1 {
-			buf, err := p.AsSerializedBuffer(t.table.schema)
-			if err != nil {
-				ascendErr = err
-				return false
-			}
-
-			// Use the dynamic row group merge adapter for each row group to ensure the missing dynamic columns are written
-			for i := 0; i < buf.NumRowGroups(); i++ {
-				rows := dynparquet.NewDynamicRowGroupMergeAdapter(ps, cols, dynCols, buf.DynamicRowGroup(i)).Rows()
-				defer rows.Close()
-
-				// Write the merged row groups to the writer
-				if _, err := rowWriter.writeRows(rows); err != nil {
-					ascendErr = err
-					return false
-				}
-			}
-		}
-
-		return true
-	})
-	if ascendErr != nil {
-		return ascendErr
-	}
+	return t.index.Rotate(t.index.MaxLevel(), t.table.externalParquetCompaction(writer))
+}
 
-	return nil
+type ParquetWriter interface {
+	Flush() error
+	WriteRows([]parquet.Row) (int, error)
+	io.Closer
 }
 
 // parquetRowWriter is a stateful parquet row group writer.
 type parquetRowWriter struct {
 	schema *dynparquet.Schema
-	w      *dynparquet.PooledWriter
+	w      ParquetWriter
 
 	rowGroupSize int
 	maxNumRows   int
@@ -1228,15 +909,9 @@ type parquetRowWriter struct {
 
 type parquetRowWriterOption func(p *parquetRowWriter)
 
-func withMaxRows(max int) parquetRowWriterOption {
-	return func(p *parquetRowWriter) {
-		p.maxNumRows = max
-	}
-}
-
 // rowWriter returns a new Parquet row writer with the given dynamic columns.
 func (t *TableBlock) rowWriter(writer io.Writer, dynCols map[string][]string, options ...parquetRowWriterOption) (*parquetRowWriter, error) {
-	w, err := t.table.schema.GetWriter(writer, dynCols)
+	w, err := t.table.schema.NewSortingWriter(writer, dynCols)
 	if err != nil {
 		return nil, err
 	}
@@ -1299,7 +974,6 @@ func (p *parquetRowWriter) writeRows(rows parquet.Rows) (int, error) {
 }
 
 func (p *parquetRowWriter) close() error {
-	defer p.schema.PutWriter(p.w)
 	return p.w.Close()
 }
 
@@ -1339,11 +1013,6 @@ func (t *Table) collectRowGroups(
 	ctx, span := t.tracer.Start(ctx, "Table/collectRowGroups")
 	defer span.End()
 
-	filter, err := BooleanExpr(filterExpr)
-	if err != nil {
-		return err
-	}
-
 	// pending blocks could be uploaded to the bucket while we iterate on them.
 	// to avoid to iterate on them again while reading the block file
 	// we keep the last block timestamp to be read from the bucket and pass it to the IterateBucketBlocks() function
@@ -1355,8 +1024,14 @@ func (t *Table) collectRowGroups(
 		}
 	}()
 	for _, block := range memoryBlocks {
-		span.AddEvent("memoryBlock")
-		if err := block.RowGroupIterator(ctx, tx, filter, rowGroups); err != nil {
+		if err := block.index.Scan(ctx, "", t.schema, filterExpr, tx, func(ctx context.Context, v any) error {
+			select {
+			case rowGroups <- v:
+				return nil
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}); err != nil {
 			return err
 		}
 	}
@@ -1391,3 +1066,85 @@ func (t *Table) close() {
 	t.active.pendingWritersWg.Wait()
 	t.closing = true
 }
+
+// configureLSMLevels configures the level configs for this table.
+func (t *Table) configureLSMLevels(levels []*index.LevelConfig) []*index.LevelConfig {
+	config := make([]*index.LevelConfig, 0, len(levels))
+
+	for _, level := range levels {
+		config = append(config, &index.LevelConfig{
+			Level:   level.Level,
+			MaxSize: level.MaxSize,
+			Compact: t.parquetCompaction, // NOTE: right now we only support parquet compaction but in the future this could compact parts into larger arrow records.
+		})
+	}
+
+	// Don't set the compaction functino for the last level. As the LSM doesn't have a level after the last one.
+	config[len(levels)-1].Compact = nil
+
+	return config
+}
+
+func (t *Table) parquetCompaction(compact []*parts.Part) ([]*parts.Part, int64, int64, error) {
+	b := &bytes.Buffer{}
+	size, err := t.compactParts(b, compact)
+	if err != nil {
+		return nil, 0, 0, err
+	}
+
+	buf, err := dynparquet.ReaderFromBytes(b.Bytes())
+	if err != nil {
+		return nil, 0, 0, err
+	}
+	return []*parts.Part{parts.NewPart(0, buf)}, size, int64(b.Len()), nil
+}
+
+func (t *Table) externalParquetCompaction(writer io.Writer) func(compact []*parts.Part) (*parts.Part, int64, int64, error) {
+	return func(compact []*parts.Part) (*parts.Part, int64, int64, error) {
+		size, err := t.compactParts(writer, compact)
+		if err != nil {
+			return nil, 0, 0, err
+		}
+
+		return nil, size, 0, nil
+	}
+}
+
+// compactParts will compact the given parts into a Parquet file written to w.
+// It returns the size in bytes of the compacted parts.
+func (t *Table) compactParts(w io.Writer, compact []*parts.Part) (int64, error) {
+	bufs := []dynparquet.DynamicRowGroup{}
+	var size int64
+	for _, part := range compact {
+		size += part.Size()
+		buf, err := part.AsSerializedBuffer(t.schema)
+		if err != nil {
+			return 0, err
+		}
+		bufs = append(bufs, buf.MultiDynamicRowGroup())
+	}
+	merged, err := t.schema.MergeDynamicRowGroups(bufs)
+	if err != nil {
+		return 0, err
+	}
+	err = func() error {
+		p, err := t.active.rowWriter(w, merged.DynamicColumns())
+		if err != nil {
+			return err
+		}
+		defer p.close()
+
+		rows := merged.Rows()
+		defer rows.Close()
+		if _, err := p.writeRows(rows); err != nil {
+			return err
+		}
+
+		return nil
+	}()
+	if err != nil {
+		return 0, err
+	}
+
+	return size, nil
+}


Test patch:
diff --git a/bench_test.go b/bench_test.go
index cea49c5..77d1c48 100644
--- a/bench_test.go
+++ b/bench_test.go
@@ -220,7 +220,6 @@ func getDeterministicLabelValuePair(ctx context.Context, engine *query.LocalEngi
 
 func BenchmarkQuery(b *testing.B) {
 	b.Skip(skipReason)
-
 	ctx := context.Background()
 	c, db, err := newDBForBenchmarks(ctx, b)
 	require.NoError(b, err)
diff --git a/compaction_test.go b/compaction_test.go
deleted file mode 100644
index 1fa03ac..0000000
--- a/compaction_test.go
+++ /dev/null
@@ -1,413 +0,0 @@
-package frostdb
-
-import (
-	"context"
-	"io"
-	"testing"
-
-	"github.com/google/btree"
-	"github.com/google/uuid"
-	"github.com/parquet-go/parquet-go"
-	"github.com/stretchr/testify/require"
-
-	"github.com/polarsignals/frostdb/dynparquet"
-	"github.com/polarsignals/frostdb/parts"
-)
-
-// insertSampleRecords is the same helper function as insertSamples but it inserts arrow records instead.
-func insertSampleRecords(ctx context.Context, t *testing.T, table *Table, timestamps ...int64) uint64 {
-	t.Helper()
-	var samples dynparquet.Samples
-	samples = make([]dynparquet.Sample, 0, len(timestamps))
-	for _, ts := range timestamps {
-		samples = append(samples, dynparquet.Sample{
-			ExampleType: "ex",
-			Labels: []dynparquet.Label{
-				{Name: "label1", Value: "value1"},
-			},
-			Stacktrace: []uuid.UUID{
-				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			},
-			Timestamp: ts,
-		})
-	}
-
-	ar, err := samples.ToRecord()
-	require.NoError(t, err)
-
-	tx, err := table.InsertRecord(ctx, ar)
-	require.NoError(t, err)
-	return tx
-}
-
-func TestCompaction(t *testing.T) {
-	// expectedPart specifies the expected part data the test should verify.
-	type expectedPart struct {
-		compactionLevel parts.CompactionLevel
-		numRowGroups    int
-		numRows         int
-		// data is the expected data. Only the timestamps are verified in this
-		// test for simplicity.
-		data []int64
-	}
-	// expectedGranule specifies the expected granule data the test should
-	// verify.
-	type expectedGranule struct {
-		parts []expectedPart
-	}
-
-	const (
-		// comactCommand indicates a compaction should be performed. Note that
-		// compactions need to be specified explicitly.
-		compactCommand = -1
-		// recordGranuleSizeCommand records the granule size at that moment.
-		recordGranuleSizeCommand = -2
-		// setRecordedGranuleSizeCommand sets the granule size to the size that
-		// was recorded when recordGranuleSizeCommand was executed.
-		setRecordedGranuleSizeCommand = -3
-		// acc accumulates the following inserts into a buffer.
-		acc = -4
-		// flushAcc inserts the accumulated data into the table. Useful to
-		// create row groups larger than one row.
-		flushAcc = -5
-	)
-
-	testCases := []struct {
-		name string
-		// rgSize is the desired row group size. If unspecified, will default to
-		// 2 rows.
-		rgSize int
-		// inserts are the timestamps to insert at. Negative int64s are
-		// interpreted as commands to do something, see the const declaration
-		// above.
-		inserts  []int64
-		expected []expectedGranule
-	}{
-		{
-			name: "SimpleLevel0ToLevel1",
-			// Insert three rows.
-			inserts: []int64{1, 2, 3, compactCommand},
-			// Expect compaction into a single part with level1 compaction.
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 3},
-						},
-					},
-				},
-			},
-		},
-		{
-			// This test is the same as above, but inserts a couple more rows,
-			// expecting a second part with compaction level 0 to be created.
-			name: "AddLevel0ToLevel1",
-			// Insert three rows.
-			inserts: []int64{1, 2, 3, compactCommand, 4},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    1,
-							compactionLevel: parts.CompactionLevel0,
-							data:            []int64{4},
-						},
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 3},
-						},
-					},
-				},
-			},
-		},
-		{
-			// This test is the same as above, but adds a command to set the
-			// granule size artificially low followed by a compaction. This
-			// should trigger a split: i.e. the new part is reassigned to a new
-			// granule.
-			name: "Split",
-			inserts: []int64{
-				1, 2, 3,
-				compactCommand, recordGranuleSizeCommand,
-				4, 5,
-				setRecordedGranuleSizeCommand, compactCommand,
-			},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 3},
-						},
-					},
-				},
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    1,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{4, 5},
-						},
-					},
-				},
-			},
-		},
-		{
-			// SimpleOverlap tests out of order inserts after a compaction. The
-			// new inserts should be merged into the existing part, because
-			// otherwise a new granule could be created with overlapping data,
-			// rendering the index useless.
-			name:    "SimpleOverlap",
-			inserts: []int64{1, 2, 3, compactCommand, 2, compactCommand},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 2, 3},
-						},
-					},
-				},
-			},
-		},
-		{
-			name: "MultiOverlap",
-			inserts: []int64{
-				1, 3, // Don't insert 2 to test non-equality overlapping.
-				compactCommand,
-				4, 5, 6,
-				compactCommand,
-				7, 8, 9,
-				compactCommand,
-				5, 2, 1,
-				compactCommand,
-			},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{7, 8, 9},
-						},
-						{
-							numRowGroups:    4,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 1, 2, 3, 4, 5, 5, 6},
-						},
-					},
-				},
-			},
-		},
-		{
-			name: "SingleL0MultipleL1Overlaps",
-			inserts: []int64{
-				1, 3,
-				compactCommand,
-				4, 5, 6,
-				compactCommand,
-				acc, 2, 4, flushAcc,
-				compactCommand,
-			},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    4,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 3, 4, 4, 5, 6},
-						},
-					},
-				},
-			},
-		},
-		{
-			name: "MultipleL0ToL1",
-			inserts: []int64{
-				acc, 1, 2, 3, flushAcc,
-				acc, 4, 5, 6, flushAcc,
-				recordGranuleSizeCommand,
-				acc, 7, 8, 9, flushAcc,
-				setRecordedGranuleSizeCommand,
-				compactCommand,
-			},
-			expected: []expectedGranule{
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{1, 2, 3, 4},
-						},
-					},
-				},
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    2,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{5, 6, 7, 8},
-						},
-					},
-				},
-				{
-					[]expectedPart{
-						{
-							numRowGroups:    1,
-							compactionLevel: parts.CompactionLevel1,
-							data:            []int64{9},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	numParts := func(g *Granule) int {
-		numparts := 0
-		g.parts.Iterate(func(p *parts.Part) bool {
-			numparts++
-			return true
-		})
-		return numparts
-	}
-
-	for _, tc := range testCases {
-		f := func() func(t *testing.T) {
-			return func(t *testing.T) {
-				c, table := basicTable(t)
-				defer c.Close()
-				// Disable interval compaction for tests. These are triggered
-				// manually.
-				table.db.compactorPool.stop()
-				table.db.compactorPool = nil
-
-				table.config.RowGroupSize = 2
-				if tc.rgSize != 0 {
-					table.config.RowGroupSize = uint64(tc.rgSize)
-				}
-
-				var (
-					numInserts          int
-					accumulating        bool
-					accBuf              []int64
-					lastTx              uint64
-					recordedGranuleSize uint64
-				)
-				for _, v := range tc.inserts {
-					switch v {
-					case compactCommand:
-						table.db.Wait(lastTx)
-						require.Equal(
-							t,
-							1,
-							table.active.Index().Len(),
-							"tests assume only a single granule as input",
-						)
-						cfg := table.db.columnStore.compactionConfig
-						cfg.l1ToGranuleSizeRatio = 0.6 // use a different ratio for arrow records
-						success, err := table.active.compactGranule((table.active.Index().Min()).(*Granule), cfg)
-						require.True(t, success)
-						require.NoError(t, err)
-					case recordGranuleSizeCommand:
-						table.db.Wait(lastTx)
-						require.Equal(
-							t,
-							1,
-							table.active.Index().Len(),
-							"tests assume only a single granule as input",
-						)
-						recordedGranuleSize = (table.active.Index().Min()).(*Granule).metadata.size.Load()
-					case setRecordedGranuleSizeCommand:
-						table.db.columnStore.granuleSizeBytes = int64(recordedGranuleSize)
-					case acc:
-						accumulating = true
-					case flushAcc:
-						accumulating = false
-						lastTx = insertSampleRecords(context.Background(), t, table, accBuf...)
-						accBuf = accBuf[:0]
-						numInserts++
-					default:
-						if accumulating {
-							accBuf = append(accBuf, v)
-							continue
-						}
-						lastTx = insertSampleRecords(context.Background(), t, table, v)
-						numInserts++
-					}
-				}
-
-				require.Equal(t, len(tc.expected), table.active.Index().Len())
-				i := 0
-				table.active.Index().Ascend(func(item btree.Item) bool {
-					g := item.(*Granule)
-					expected := tc.expected[i]
-					require.Equal(t, len(expected.parts), numParts(g))
-
-					j := 0
-					g.parts.Iterate(func(p *parts.Part) bool {
-						expectedPart := expected.parts[j]
-						if expectedPart.numRowGroups == 0 {
-							require.Equal(t, int64(expectedPart.numRows), p.Record().NumRows())
-						} else {
-							buf, err := p.AsSerializedBuffer(table.Schema())
-							require.NoError(t, err)
-							rgs := buf.ParquetFile().RowGroups()
-							require.Equal(t, expectedPart.numRowGroups, len(rgs))
-							require.Equal(t, expectedPart.compactionLevel, p.CompactionLevel())
-							rowsRead := make([]parquet.Row, 0)
-							for _, rg := range rgs {
-								func() {
-									rows := rg.Rows()
-									defer rows.Close()
-
-									for {
-										rowBuf := make([]parquet.Row, 1)
-										n, err := rows.ReadRows(rowBuf)
-										if err != nil && err != io.EOF {
-											require.NoError(t, err)
-										}
-										if n > 0 {
-											rowsRead = append(rowsRead, rowBuf...)
-										}
-
-										if err == io.EOF {
-											break
-										}
-									}
-								}()
-							}
-							require.Equal(
-								t,
-								len(expectedPart.data),
-								len(rowsRead),
-								"different number of rows read for granule %d part %d",
-								i,
-								j,
-							)
-
-							// This is a bit of a hack. If the check below fails
-							// unexpectedly after a change to the default schema, think
-							// about a more robust search of the timestamp column index.
-							const timestampColumnIdx = 3
-							for k, expectedTimestamp := range expectedPart.data {
-								require.Equal(t, rowsRead[k][timestampColumnIdx].Int64(), expectedTimestamp)
-							}
-						}
-
-						j++
-						return true
-					})
-					i++
-					return true
-				})
-			}
-		}
-		t.Run(tc.name+"-arrow", f())
-	}
-}
diff --git a/db_test.go b/db_test.go
index 82aaf29..918580e 100644
--- a/db_test.go
+++ b/db_test.go
@@ -74,6 +74,28 @@ func TestDBWithWALAndBucket(t *testing.T) {
 	)
 	require.NoError(t, err)
 	defer c.Close()
+	db, err = c.DB(context.Background(), "test")
+	require.NoError(t, err)
+	table, err = db.Table("test", config)
+	require.NoError(t, err)
+
+	// Validate that a read can be performed of the persisted data
+	pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+	defer pool.AssertSize(t, 0)
+	rows := int64(0)
+	err = table.View(ctx, func(ctx context.Context, tx uint64) error {
+		return table.Iterator(
+			ctx,
+			tx,
+			pool,
+			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
+				rows += ar.NumRows()
+				return nil
+			}},
+		)
+	})
+	require.NoError(t, err)
+	require.Equal(t, int64(300), rows)
 }
 
 func TestDBWithWAL(t *testing.T) {
diff --git a/index/lsm_test.go b/index/lsm_test.go
new file mode 100644
index 0000000..a3c84da
--- /dev/null
+++ b/index/lsm_test.go
@@ -0,0 +1,208 @@
+package index
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/apache/arrow/go/v14/arrow"
+	"github.com/parquet-go/parquet-go"
+	"github.com/stretchr/testify/require"
+
+	"github.com/polarsignals/frostdb/dynparquet"
+	"github.com/polarsignals/frostdb/parts"
+)
+
+func parquetCompaction(compact []*parts.Part) ([]*parts.Part, int64, int64, error) {
+	b := &bytes.Buffer{}
+	size, err := compactParts(b, compact)
+	if err != nil {
+		return nil, 0, 0, err
+	}
+
+	buf, err := dynparquet.ReaderFromBytes(b.Bytes())
+	if err != nil {
+		return nil, 0, 0, err
+	}
+	return []*parts.Part{parts.NewPart(0, buf)}, size, int64(b.Len()), nil
+}
+
+func compactParts(w io.Writer, compact []*parts.Part) (int64, error) {
+	schema := dynparquet.NewSampleSchema()
+	bufs := []dynparquet.DynamicRowGroup{}
+	var size int64
+	for _, part := range compact {
+		size += part.Size()
+		buf, err := part.AsSerializedBuffer(schema)
+		if err != nil {
+			return 0, err
+		}
+		bufs = append(bufs, buf.MultiDynamicRowGroup())
+	}
+	merged, err := schema.MergeDynamicRowGroups(bufs)
+	if err != nil {
+		return 0, err
+	}
+	err = func() error {
+		writer, err := schema.GetWriter(w, merged.DynamicColumns())
+		if err != nil {
+			return err
+		}
+		defer writer.Close()
+
+		rows := merged.Rows()
+		defer rows.Close()
+
+		buf := make([]parquet.Row, merged.NumRows())
+		if _, err := rows.ReadRows(buf); err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+		if _, err := writer.WriteRows(buf); err != nil && !errors.Is(err, io.EOF) {
+			return err
+		}
+
+		return nil
+	}()
+	if err != nil {
+		return 0, err
+	}
+
+	return size, nil
+}
+
+func check(t *testing.T, lsm *LSM, records, buffers int) {
+	t.Helper()
+	seen := map[SentinelType]bool{}
+	lsm.levels.Iterate(func(node *Node) bool {
+		if node.part == nil {
+			if seen[node.sentinel] {
+				t.Fatal("duplicate sentinel")
+			}
+			seen[node.sentinel] = true
+		}
+		return true
+	})
+	rec := 0
+	buf := 0
+	require.NoError(t, lsm.Scan(context.Background(), "", nil, nil, 2, func(ctx context.Context, v any) error {
+		switch v.(type) {
+		case arrow.Record:
+			rec++
+		case dynparquet.DynamicRowGroup:
+			buf++
+		}
+		return nil
+	}))
+	require.Equal(t, records, rec)
+	require.Equal(t, buf, buffers)
+}
+
+func Test_LSM_Basic(t *testing.T) {
+	t.Parallel()
+	lsm, err := NewLSM("test", []*LevelConfig{
+		{Level: L0, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L1, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L2, MaxSize: 1024 * 1024 * 1024},
+	})
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	lsm.Add(1, r)
+	lsm.Add(1, r)
+	lsm.Add(1, r)
+	check(t, lsm, 3, 0)
+	require.NoError(t, lsm.merge(L0, nil))
+	check(t, lsm, 0, 1)
+	lsm.Add(1, r)
+	check(t, lsm, 1, 1)
+	lsm.Add(1, r)
+	check(t, lsm, 2, 1)
+	require.NoError(t, lsm.merge(L0, nil))
+	check(t, lsm, 0, 2)
+	lsm.Add(1, r)
+	check(t, lsm, 1, 2)
+	require.NoError(t, lsm.merge(L1, nil))
+	check(t, lsm, 1, 1)
+	require.NoError(t, lsm.merge(L0, nil))
+	check(t, lsm, 0, 2)
+}
+
+func Test_LSM_DuplicateSentinel(t *testing.T) {
+	t.Parallel()
+	lsm, err := NewLSM("test", []*LevelConfig{
+		{Level: L0, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L1, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L2, MaxSize: 1024 * 1024 * 1024},
+	})
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	lsm.Add(1, r)
+	lsm.Add(1, r)
+	lsm.Add(1, r)
+	check(t, lsm, 3, 0)
+	require.NoError(t, lsm.merge(L0, nil))
+	check(t, lsm, 0, 1)
+	require.NoError(t, lsm.merge(L0, nil))
+	check(t, lsm, 0, 1)
+}
+
+func Test_LSM_Compaction(t *testing.T) {
+	t.Parallel()
+	lsm, err := NewLSM("test", []*LevelConfig{
+		{Level: L0, MaxSize: 1, Compact: parquetCompaction},
+		{Level: L1, MaxSize: 1024 * 1024 * 1024},
+	})
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	lsm.Add(1, r)
+	require.Eventually(t, func() bool {
+		return lsm.sizes[L0].Load() == 0 && lsm.sizes[L1].Load() != 0
+	}, time.Second, 5*time.Millisecond)
+}
+
+func Test_LSM_CascadeCompaction(t *testing.T) {
+	t.Parallel()
+	lsm, err := NewLSM("test", []*LevelConfig{
+		{Level: L0, MaxSize: 257, Compact: parquetCompaction},
+		{Level: L1, MaxSize: 2281, Compact: parquetCompaction},
+		{Level: L2, MaxSize: 2281, Compact: parquetCompaction},
+		{Level: 3, MaxSize: 2281, Compact: parquetCompaction},
+		{Level: 4, MaxSize: 2281},
+	})
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	lsm.Add(1, r)
+	require.Eventually(t, func() bool {
+		return lsm.sizes[L0].Load() == 0 &&
+			lsm.sizes[L1].Load() != 0 &&
+			lsm.sizes[L2].Load() == 0 &&
+			lsm.sizes[3].Load() == 0 &&
+			lsm.sizes[4].Load() == 0
+	}, time.Second, 5*time.Millisecond)
+	lsm.Add(1, r)
+	require.Eventually(t, func() bool {
+		return lsm.sizes[L0].Load() == 0 &&
+			lsm.sizes[L1].Load() == 0 &&
+			lsm.sizes[L2].Load() == 0 &&
+			lsm.sizes[3].Load() == 0 &&
+			lsm.sizes[4].Load() != 0
+	}, time.Second, 5*time.Millisecond)
+}
diff --git a/binaryscalarexpr_test.go b/query/expr/binaryscalarexpr_test.go
similarity index 99%
rename from binaryscalarexpr_test.go
rename to query/expr/binaryscalarexpr_test.go
index 85309bb..570030f 100644
--- a/binaryscalarexpr_test.go
+++ b/query/expr/binaryscalarexpr_test.go
@@ -1,4 +1,4 @@
-package frostdb
+package expr
 
 import (
 	"testing"
diff --git a/snapshot_test.go b/snapshot_test.go
index b6b472a..dcf6584 100644
--- a/snapshot_test.go
+++ b/snapshot_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
 	"github.com/apache/arrow/go/v14/arrow/memory"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 
@@ -20,6 +21,32 @@ import (
 	"github.com/polarsignals/frostdb/query/logicalplan"
 )
 
+// insertSampleRecords is the same helper function as insertSamples but it inserts arrow records instead.
+func insertSampleRecords(ctx context.Context, t *testing.T, table *Table, timestamps ...int64) uint64 {
+	t.Helper()
+	var samples dynparquet.Samples
+	samples = make([]dynparquet.Sample, 0, len(timestamps))
+	for _, ts := range timestamps {
+		samples = append(samples, dynparquet.Sample{
+			ExampleType: "ex",
+			Labels: []dynparquet.Label{
+				{Name: "label1", Value: "value1"},
+			},
+			Stacktrace: []uuid.UUID{
+				{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
+			},
+			Timestamp: ts,
+		})
+	}
+
+	ar, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	tx, err := table.InsertRecord(ctx, ar)
+	require.NoError(t, err)
+	return tx
+}
+
 func TestSnapshot(t *testing.T) {
 	ctx := context.Background()
 	// Create a new DB with multiple tables and granules with
@@ -64,8 +91,6 @@ func TestSnapshot(t *testing.T) {
 
 		config := NewTableConfig(dynparquet.SampleDefinition())
 
-		// Pause compactor pool to have control over compactions.
-		db.compactorPool.pause()
 		table, err := db.Table("table1", config)
 		require.NoError(t, err)
 		insertSampleRecords(ctx, t, table, 1, 2, 3)
@@ -94,7 +119,6 @@ func TestSnapshot(t *testing.T) {
 		snapshotDB, err := c.DB(ctx, "testsnapshot")
 		require.NoError(t, err)
 
-		snapshotDB.compactorPool.pause()
 		// Load the other db's latest snapshot.
 		tx, err := snapshotDB.loadLatestSnapshotFromDir(ctx, db.snapshotsDir())
 		require.NoError(t, err)
diff --git a/table_test.go b/table_test.go
index de3b58b..328321e 100644
--- a/table_test.go
+++ b/table_test.go
@@ -10,6 +10,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
@@ -19,11 +20,11 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/google/uuid"
 	"github.com/parquet-go/parquet-go"
-	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 
 	"github.com/polarsignals/frostdb/dynparquet"
 	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
+	"github.com/polarsignals/frostdb/index"
 	"github.com/polarsignals/frostdb/pqarrow"
 	"github.com/polarsignals/frostdb/query"
 	"github.com/polarsignals/frostdb/query/logicalplan"
@@ -71,132 +72,6 @@ func basicTable(t *testing.T, options ...Option) (*ColumnStore, *Table) {
 	return c, table
 }
 
-func TestTable(t *testing.T) {
-	c, table := basicTable(t)
-	defer c.Close()
-
-	samples := dynparquet.Samples{{
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value1"},
-			{Name: "label2", Value: "value2"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 1,
-		Value:     1,
-	}, {
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value2"},
-			{Name: "label2", Value: "value2"},
-			{Name: "label3", Value: "value3"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 2,
-		Value:     2,
-	}, {
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value3"},
-			{Name: "label2", Value: "value2"},
-			{Name: "label4", Value: "value4"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 3,
-		Value:     3,
-	}}
-
-	r, err := samples.ToRecord()
-	require.NoError(t, err)
-
-	ctx := context.Background()
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
-
-	samples = dynparquet.Samples{{
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value1"},
-			{Name: "label2", Value: "value2"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 2,
-		Value:     2,
-	}}
-
-	r, err = samples.ToRecord()
-	require.NoError(t, err)
-
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
-
-	samples = dynparquet.Samples{{
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value1"},
-			{Name: "label2", Value: "value2"},
-			{Name: "label3", Value: "value3"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 3,
-		Value:     3,
-	}}
-
-	r, err = samples.ToRecord()
-	require.NoError(t, err)
-
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
-
-	pool := memory.NewGoAllocator()
-
-	err = table.View(ctx, func(ctx context.Context, tx uint64) error {
-		return table.Iterator(
-			ctx,
-			tx,
-			pool,
-			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
-				t.Log(ar)
-				return nil
-			}},
-		)
-	})
-	require.NoError(t, err)
-
-	uuid1 := uuid.MustParse("00000000-0000-0000-0000-000000000001")
-	uuid2 := uuid.MustParse("00000000-0000-0000-0000-000000000002")
-
-	// One granule with 3 parts
-	require.Equal(t, 1, table.active.Index().Len())
-	require.Equal(t, 3, table.active.Index().Min().(*Granule).parts.Total())
-	require.Equal(t, parquet.Row{
-		parquet.ValueOf("test").Level(0, 0, 0),
-		parquet.ValueOf("value1").Level(0, 1, 1),
-		parquet.ValueOf("value2").Level(0, 1, 2),
-		parquet.ValueOf(nil).Level(0, 0, 3),
-		parquet.ValueOf(nil).Level(0, 0, 4),
-		parquet.ValueOf(append(uuid1[:], uuid2[:]...)).Level(0, 0, 5),
-		parquet.ValueOf(1).Level(0, 0, 6),
-		parquet.ValueOf(1).Level(0, 0, 7),
-	}, (*dynparquet.DynamicRow)(table.active.Index().Min().(*Granule).metadata.least).Row)
-	require.Equal(t, 1, table.active.Index().Len())
-}
-
 // This test issues concurrent writes to the database, and expects all of them to be recorded successfully.
 func Test_Table_Concurrency(t *testing.T) {
 	tests := map[string]struct {
@@ -399,7 +274,6 @@ func benchmarkTableInserts(b *testing.B, rows, iterations, writers int) {
 				}},
 			)
 		})
-		require.Equal(b, 0., testutil.ToFloat64(table.metrics.granulesCompactionAborted))
 		require.NoError(b, err)
 		require.Equal(b, int64(rows*iterations*writers), totalrows)
 
@@ -576,7 +450,12 @@ func Test_Table_NewTableValidSplitSize(t *testing.T) {
 }
 
 func Test_Table_Bloomfilter(t *testing.T) {
-	c, table := basicTable(t, WithGranuleSizeBytes(1))
+	c, table := basicTable(t, WithIndexConfig(
+		[]*index.LevelConfig{
+			{Level: index.L0, MaxSize: 452}, // NOTE: 452 is the current size of the 3 records that are inserted
+			{Level: index.L1, MaxSize: 100000},
+		},
+	))
 	defer c.Close()
 
 	samples := dynparquet.Samples{{
@@ -631,24 +510,26 @@ func Test_Table_Bloomfilter(t *testing.T) {
 
 	require.NoError(t, table.EnsureCompaction())
 
-	iterations := 0
-	err := table.View(context.Background(), func(ctx context.Context, tx uint64) error {
-		pool := memory.NewGoAllocator()
+	require.Eventually(t, func() bool {
+		iterations := 0
+		err := table.View(context.Background(), func(ctx context.Context, tx uint64) error {
+			pool := memory.NewGoAllocator()
 
-		require.NoError(t, table.Iterator(
-			context.Background(),
-			tx,
-			pool,
-			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
-				iterations++
-				return nil
-			}},
-			logicalplan.WithFilter(logicalplan.Col("labels.label4").Eq(logicalplan.Literal("value4"))),
-		))
-		return nil
-	})
-	require.NoError(t, err)
-	require.Equal(t, 1, iterations)
+			require.NoError(t, table.Iterator(
+				context.Background(),
+				tx,
+				pool,
+				[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
+					iterations++
+					return nil
+				}},
+				logicalplan.WithFilter(logicalplan.Col("labels.label4").Eq(logicalplan.Literal("value4"))),
+			))
+			return nil
+		})
+		require.NoError(t, err)
+		return iterations == 1
+	}, time.Millisecond*60, time.Millisecond*10)
 }
 
 func Test_RecordToRow(t *testing.T) {
@@ -766,82 +647,6 @@ func Test_L0Query(t *testing.T) {
 	require.Equal(t, 1, records)
 }
 
-// This test checks to make sure that if a new row is added that is globally the least it shall be added as a new granule.
-func Test_Table_InsertLeast(t *testing.T) {
-	c, table := basicTable(t)
-	defer c.Close()
-
-	samples := dynparquet.Samples{{
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value1"},
-			{Name: "label2", Value: "value2"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 1,
-		Value:     1,
-	}, {
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value2"},
-			{Name: "label2", Value: "value2"},
-			{Name: "label3", Value: "value3"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 2,
-		Value:     2,
-	}, {
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "value3"},
-			{Name: "label2", Value: "value2"},
-			{Name: "label4", Value: "value4"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 3,
-		Value:     3,
-	}}
-
-	r, err := samples.ToRecord()
-	require.NoError(t, err)
-
-	ctx := context.Background()
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
-
-	before := table.active.Index().Len()
-
-	samples = dynparquet.Samples{{
-		ExampleType: "test",
-		Labels: []dynparquet.Label{
-			{Name: "label1", Value: "a"},
-		},
-		Stacktrace: []uuid.UUID{
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1},
-			{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
-		},
-		Timestamp: 2,
-		Value:     2,
-	}}
-
-	r, err = samples.ToRecord()
-	require.NoError(t, err)
-
-	_, err = table.InsertRecord(ctx, r)
-	require.NoError(t, err)
-
-	require.Equal(t, before+1, table.active.Index().Len())
-}
-
 func Test_Serialize_DisparateDynamicColumns(t *testing.T) {
 	c, table := basicTable(t)
 	defer c.Close()
@@ -1328,8 +1133,10 @@ func Test_Table_DynamicColumnNotDefined(t *testing.T) {
 }
 
 func TestTableUniquePrimaryIndex(t *testing.T) {
-	// Disable compaction so we trigger it manually.
-	c, err := New(WithGranuleSizeBytes(1), WithCompactionConfig(NewCompactionConfig(WithConcurrency(0))))
+	c, err := New(WithGranuleSizeBytes(1), WithIndexConfig([]*index.LevelConfig{
+		{Level: index.L0, MaxSize: 180},
+		{Level: index.L1, MaxSize: 1024 * 1024 * 1024 * 1024},
+	}))
 	require.NoError(t, err)
 	defer c.Close()
 
@@ -1359,7 +1166,7 @@ func TestTableUniquePrimaryIndex(t *testing.T) {
 	table, err := db.Table(tableName, NewTableConfig(schema))
 	require.NoError(t, err)
 
-	const numRecords = 10
+	const numRecords = 9
 	for i := 0; i < numRecords; i++ {
 		_, err = table.Write(context.Background(), struct {
 			Name string
@@ -1380,6 +1187,13 @@ func TestTableUniquePrimaryIndex(t *testing.T) {
 	// Duplicates are only dropped after compaction.
 	require.Equal(t, numRecords, rowsRead)
 
+	// Trigger compaction with a new record.
+	_, err = table.Write(context.Background(), struct {
+		Name string
+	}{
+		Name: "duplicate",
+	})
+	require.NoError(t, err)
 	require.NoError(t, table.ActiveBlock().EnsureCompaction())
 
 	rowsRead = 0