Repository: polarsignals/frostdb
Pull request #667 (2024-01-02 19:20:49)

Persistent lsm
Adds support for persistent LSM files.

If storage path is enabled then persistent LSM files are used instead of temporary files. During startup, the database will recover parts from the LSM files and insert them into the index into the corresponding levels. 

If persistent LSM is configured then snapshots will not write those compacted parts into the snapshot file, and will only write the L0 parts.


Patch:
diff --git a/compaction.go b/compaction.go
new file mode 100644
index 0000000..37141a3
--- /dev/null
+++ b/compaction.go
@@ -0,0 +1,209 @@
+package frostdb
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/go-kit/log/level"
+	"github.com/oklog/ulid"
+	"github.com/parquet-go/parquet-go"
+
+	"github.com/polarsignals/frostdb/dynparquet"
+	"github.com/polarsignals/frostdb/parts"
+)
+
+const (
+	ParquetCompactionTXKey = "compaction_tx"
+)
+
+var ErrCompactionRecoveryFailed = fmt.Errorf("failed to recover compacted level")
+
+type FileCompaction struct {
+	t      *Table
+	file   *os.File
+	offset int64 // Writing offsets into the file
+	ref    int64 // Number of references to file.
+}
+
+func (f *FileCompaction) Close(cleanup bool) error {
+	err := f.file.Close()
+	if cleanup {
+		if err := os.Remove(f.file.Name()); err != nil {
+			return fmt.Errorf("failed to remove file: %v", err)
+		}
+	}
+	return err
+}
+
+func NewFileCompaction(t *Table, block ulid.ULID, lvl int) *FileCompaction {
+	f := &FileCompaction{
+		t: t,
+	}
+
+	if t.db.storagePath == "" { // No storage path provided, use temp files.
+		file, err := os.CreateTemp("", fmt.Sprintf("L%v-*.parquet", lvl))
+		if err != nil {
+			panic(err)
+		}
+		level.Info(t.logger).Log("msg", "created temp file for level", "level", lvl, "file", file.Name())
+		f.file = file
+		return f
+	}
+
+	indexDir := filepath.Join(t.db.indexDir(), t.name)
+	if err := os.MkdirAll(indexDir, dirPerms); err != nil {
+		panic(err)
+	}
+
+	file, err := os.OpenFile(filepath.Join(indexDir, fmt.Sprintf("L%v-%s.parquet", lvl, block.String())), os.O_CREATE|os.O_RDWR, filePerms)
+	if err != nil {
+		panic(err)
+	}
+	f.file = file
+	return f
+}
+
+// accountingWriter is a writer that accounts for the number of bytes written.
+type accountingWriter struct {
+	w io.Writer
+	n int64
+}
+
+func (a *accountingWriter) Write(p []byte) (int, error) {
+	n, err := a.w.Write(p)
+	a.n += int64(n)
+	return n, err
+}
+
+// writeRecordsToParquetFile will compact the given parts into a Parquet file written to the next level file.
+func (f *FileCompaction) writeRecordsToParquetFile(compact []parts.Part, options ...parts.Option) ([]parts.Part, int64, int64, error) {
+	if len(compact) == 0 {
+		return nil, 0, 0, fmt.Errorf("no parts to compact")
+	}
+
+	accountant := &accountingWriter{w: f.file}
+	preCompactionSize, err := f.t.compactParts(accountant, compact,
+		parquet.KeyValueMetadata(
+			ParquetCompactionTXKey, // Compacting up through this transaction.
+			fmt.Sprintf("%v", compact[0].TX()),
+		),
+	) // compact into the next level
+	if err != nil {
+		return nil, 0, 0, err
+	}
+
+	// Record the writing offset into the file.
+	prevOffset := f.offset
+	f.offset += accountant.n + 8
+	f.ref++
+
+	// Record the file size for recovery.
+	size := make([]byte, 8)
+	binary.LittleEndian.PutUint64(size, uint64(accountant.n))
+	if n, err := f.file.Write(size); n != 8 {
+		return nil, 0, 0, fmt.Errorf("failed to write size to file: %v", err)
+	}
+
+	pf, err := parquet.OpenFile(io.NewSectionReader(f.file, prevOffset, accountant.n), accountant.n)
+	if err != nil {
+		return nil, 0, 0, err
+	}
+
+	buf, err := dynparquet.NewSerializedBuffer(pf)
+	if err != nil {
+		return nil, 0, 0, err
+	}
+
+	return []parts.Part{parts.NewParquetPart(compact[0].TX(), buf, append(options, parts.WithRelease(f.release()))...)}, preCompactionSize, accountant.n, nil
+}
+
+// Truncate will truncate the file to 0 bytes. This is used when a compaction recovery fails.
+func (f *FileCompaction) Truncate() error {
+	return f.file.Truncate(0)
+}
+
+// release will account for all the Parts currently pointing to this file. Once the last one has been released it will truncate the file.
+// release is not safe to call concurrently.
+func (f *FileCompaction) release() func() {
+	return func() {
+		f.ref--
+		if f.ref == 0 {
+			level.Info(f.t.logger).Log("msg", "truncating file", "file", f.file.Name())
+			if err := os.Truncate(f.file.Name(), 0); err != nil {
+				panic(fmt.Errorf("failed to truncate the level file: %v", err))
+			}
+
+			f.offset = 0
+			if _, err := f.file.Seek(0, io.SeekStart); err != nil {
+				panic(fmt.Errorf("failed to seek the level file: %v", err))
+			}
+		}
+	}
+}
+
+func (f *FileCompaction) recover(options ...parts.Option) ([]parts.Part, error) {
+	recovered, err := func() ([]parts.Part, error) {
+		info, err := os.Stat(f.file.Name())
+		if err != nil {
+			return nil, err
+		}
+
+		if info.Size() == 0 { // file was truncated, nothing to recover.
+			return nil, nil
+		}
+
+		recovered := []parts.Part{}
+
+		// Recover all parts from file.
+		for offset := info.Size(); offset > 0; {
+			offset -= 8
+			size := make([]byte, 8)
+			if n, err := f.file.ReadAt(size, offset); n != 8 {
+				return nil, fmt.Errorf("failed to read size from file: %v", err)
+			}
+			parquetSize := int64(binary.LittleEndian.Uint64(size))
+			offset -= parquetSize
+
+			pf, err := parquet.OpenFile(io.NewSectionReader(f.file, offset, parquetSize), parquetSize)
+			if err != nil {
+				return nil, err
+			}
+
+			buf, err := dynparquet.NewSerializedBuffer(pf)
+			if err != nil {
+				return nil, err
+			}
+
+			txstr, ok := buf.ParquetFile().Lookup(ParquetCompactionTXKey)
+			if !ok {
+				return nil, fmt.Errorf("failed to find compaction_tx metadata")
+			}
+
+			tx, err := strconv.Atoi(txstr)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse compaction_tx metadata: %v", err)
+			}
+
+			recovered = append(recovered, parts.NewParquetPart(uint64(tx), buf, append(options, parts.WithRelease(f.release()))...))
+		}
+
+		return recovered, nil
+	}()
+	if err != nil {
+		level.Error(f.t.logger).Log("msg", "truncating file after failed recovery", "err", err, "file", f.file.Name())
+		if err := os.Truncate(f.file.Name(), 0); err != nil {
+			return nil, fmt.Errorf("failed to truncate the level file %s: %v", f.file.Name(), err)
+		}
+
+		return nil, ErrCompactionRecoveryFailed
+	}
+
+	return recovered, err
+}
+
+// Sync calls Sync on the underlying file.
+func (f *FileCompaction) Sync() error { return f.file.Sync() }
diff --git a/db.go b/db.go
index 17b49c1..55bcc8a 100644
--- a/db.go
+++ b/db.go
@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/apache/arrow/go/v14/arrow/ipc"
+	"github.com/apache/arrow/go/v14/arrow/util"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid"
@@ -32,6 +33,7 @@ import (
 	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/query/logicalplan"
 	"github.com/polarsignals/frostdb/wal"
 )
@@ -128,6 +130,14 @@ func New(
 		return nil, fmt.Errorf("storage path must be configured if WAL is enabled")
 	}
 
+	for _, cfg := range s.indexConfig {
+		if cfg.Type == CompactionTypeParquetDisk {
+			if !s.enableWAL && s.storagePath != "" {
+				return nil, fmt.Errorf("persistent disk compaction requires WAL to be enabled")
+			}
+		}
+	}
+
 	if err := s.recoverDBsFromStorage(context.Background()); err != nil {
 		return nil, err
 	}
@@ -481,12 +491,15 @@ func (s *ColumnStore) DB(ctx context.Context, name string, opts ...DBOption) (*D
 		reg:         reg,
 		logger:      logger,
 		tracer:      s.tracer,
-		storagePath: filepath.Join(s.DatabasesDir(), name),
 		wal:         &wal.NopWAL{},
 		sources:     s.sources,
 		sinks:       s.sinks,
 	}
 
+	if s.storagePath != "" {
+		db.storagePath = filepath.Join(s.DatabasesDir(), name)
+	}
+
 	if err := applyOptsToDB(db); err != nil {
 		return nil, err
 	}
@@ -669,8 +682,17 @@ func (db *DB) trashDir() string {
 	return filepath.Join(db.storagePath, "trash")
 }
 
-// recover attempts to recover database state from a combination of snapshots
-// and the WAL.
+func (db *DB) indexDir() string {
+	return filepath.Join(db.storagePath, "index")
+}
+
+// recover attempts to recover database state from a combination of snapshots and the WAL.
+//
+// The recovery process is as follows:
+// 1. Load the latest snapshot (if one should exist).
+// 1.a. If on-disk LSM index files exist: Upon table creation during snapshot loading, the index files shall be recovered from, inserting parts into the index.
+// 2. Replay the WAL starting from the latest snapshot transaction.
+// 2.a. If on-disk LSM index files were loaded: Insertion into the index may drop the insertion if a part with a higher transaction already exists in the WAL.
 func (db *DB) recover(ctx context.Context, wal WAL) error {
 	level.Info(db.logger).Log(
 		"msg", "recovering db",
@@ -727,14 +749,24 @@ 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 {
+					cfg, recovered, err := table.ActiveBlock().configureLSMLevels(db.columnStore.indexConfig)
+					if err != nil {
+						return fmt.Errorf("configure lsm levels: %w", err)
+					}
+
 					table.ActiveBlock().index, err = index.NewLSM(
 						table.name,
 						table.schema,
-						table.configureLSMLevels(db.columnStore.indexConfig),
+						cfg,
+						db.Wait,
 						index.LSMWithMetrics(table.metrics.indexMetrics),
 					)
 					if err != nil {
-						return err
+						return fmt.Errorf("create new lsm index: %w", err)
+					}
+
+					for _, p := range recovered {
+						table.ActiveBlock().index.InsertPart(p)
 					}
 				}
 				db.mtx.Unlock()
@@ -752,9 +784,6 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 	// WAL (i.e. entries that occupy space on disk but are useless).
 	performSnapshot := false
 
-	// Writes are performed concurrently to speed up replay.
-	var writeWg errgroup.Group
-	writeWg.SetLimit(runtime.GOMAXPROCS(0))
 	if err := wal.Replay(snapshotTx, func(tx uint64, record *walpb.Record) error {
 		if err := ctx.Err(); err != nil {
 			return err
@@ -878,27 +907,26 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 				return fmt.Errorf("get table: %w", err)
 			}
 
-			writeWg.Go(func() error {
-				switch e.Write.Arrow {
-				case true:
-					reader, err := ipc.NewReader(bytes.NewReader(entry.Data))
-					if err != nil {
-						return fmt.Errorf("create ipc reader: %w", err)
-					}
-					record, err := reader.Read()
-					if err != nil {
-						return fmt.Errorf("read record: %w", err)
-					}
-
-					if err := table.active.InsertRecord(ctx, tx, record); err != nil {
-						return fmt.Errorf("insert record into block: %w", err)
-					}
-				default:
-					panic("parquet writes are deprecated")
+			switch e.Write.Arrow {
+			case true:
+				reader, err := ipc.NewReader(bytes.NewReader(entry.Data))
+				if err != nil {
+					return fmt.Errorf("create ipc reader: %w", err)
 				}
-				return nil
-			})
+				record, err := reader.Read()
+				if err != nil {
+					return fmt.Errorf("read record: %w", err)
+				}
+				defer reader.Release()
 
+				// TODO: the issue with this is we aren't prehashing the columns by not doing InsertRecord.
+				// If we prehash before WAL write we don't have to do that here.
+				size := util.TotalRecordSize(record)
+				table.active.index.InsertPart(parts.NewArrowPart(tx, record, uint64(size), table.schema, parts.WithCompactionLevel(int(index.L0))))
+			default:
+				panic("parquet writes are deprecated")
+			}
+			return nil
 		case *walpb.Entry_TableBlockPersisted_:
 			// If a block was persisted but the entry still exists in the WAL,
 			// a snapshot was not performed after persisting the block. Perform
@@ -915,10 +943,6 @@ func (db *DB) recover(ctx context.Context, wal WAL) error {
 		return err
 	}
 
-	if err := writeWg.Wait(); err != nil {
-		return fmt.Errorf("write error during replay: %w", err)
-	}
-
 	resetTxn := snapshotTx
 	if lastTx > resetTxn {
 		resetTxn = lastTx
@@ -1008,7 +1032,7 @@ func (db *DB) Close(options ...CloseOption) error {
 		return err
 	}
 
-	if shouldPersist || opts.clearStorage {
+	if (shouldPersist || opts.clearStorage) && db.storagePath != "" {
 		if err := db.dropStorage(); err != nil {
 			return err
 		}
@@ -1121,6 +1145,10 @@ func (db *DB) promoteReadOnlyTableLocked(name string, config *tablepb.TableConfi
 
 // Table will get or create a new table with the given name and config. If a table already exists with the given name, it will have it's configuration updated.
 func (db *DB) Table(name string, config *tablepb.TableConfig) (*Table, error) {
+	return db.table(name, config, generateULID())
+}
+
+func (db *DB) table(name string, config *tablepb.TableConfig, id ulid.ULID) (*Table, error) {
 	if config == nil {
 		return nil, fmt.Errorf("table config cannot be nil")
 	}
@@ -1171,7 +1199,6 @@ func (db *DB) Table(name string, config *tablepb.TableConfig) (*Table, error) {
 	tx, _, commit := db.begin()
 	defer commit()
 
-	id := generateULID()
 	if err := table.newTableBlock(0, tx, id); err != nil {
 		return nil, err
 	}
@@ -1286,6 +1313,18 @@ func (db *DB) resetToTxn(txn uint64, wal WAL) {
 	db.tx.Store(txn)
 	db.highWatermark.Store(txn)
 	if wal != nil {
+		// Before resetting the wal make sure any pending writes to the index are flushed.
+		errg := errgroup.Group{}
+		for _, table := range db.tables {
+			for _, syncer := range table.ActiveBlock().syncers {
+				errg.Go(syncer.Sync)
+			}
+		}
+
+		if err := errg.Wait(); err != nil {
+			level.Error(db.logger).Log("msg", "failed to sync index before resetting WAL; dataloss may occur", "err", err)
+		}
+
 		// This call resets the WAL to a zero state so that new records can be
 		// logged.
 		if err := wal.Reset(txn + 1); err != nil {
diff --git a/dynparquet/schema.go b/dynparquet/schema.go
index 570d6ee..9d01a1f 100644
--- a/dynparquet/schema.go
+++ b/dynparquet/schema.go
@@ -1072,7 +1072,7 @@ const bloomFilterBitsPerValue = 10
 
 // NewWriter returns a new parquet writer with a concrete parquet schema
 // generated using the given concrete dynamic column names.
-func (s *Schema) NewWriter(w io.Writer, dynamicColumns map[string][]string, sorting bool) (ParquetWriter, error) {
+func (s *Schema) NewWriter(w io.Writer, dynamicColumns map[string][]string, sorting bool, options ...parquet.WriterOption) (ParquetWriter, error) {
 	ps, err := s.GetDynamicParquetSchema(dynamicColumns)
 	if err != nil {
 		return nil, err
@@ -1109,6 +1109,7 @@ func (s *Schema) NewWriter(w io.Writer, dynamicColumns map[string][]string, sort
 			parquet.SortingColumns(cols...),
 		),
 	}
+	writerOptions = append(writerOptions, options...)
 	if sorting {
 		return parquet.NewSortingWriter[any](w, 32*1024, writerOptions...), nil
 	}
diff --git a/index/lsm.go b/index/lsm.go
index 80e9496..18004c4 100644
--- a/index/lsm.go
+++ b/index/lsm.go
@@ -33,13 +33,16 @@ type LSM struct {
 
 	schema *dynparquet.Schema
 
-	prefix  string
-	levels  *Node
-	sizes   []atomic.Int64
-	configs []*LevelConfig
+	prefix        string
+	maxTXRecoverd []uint64
+	levels        *Node
+	sizes         []atomic.Int64
+	configs       []*LevelConfig
 
+	// Options
 	logger  log.Logger
 	metrics *LSMMetrics
+	wait    func(uint64)
 }
 
 // LSMMetrics are the metrics for an LSM index.
@@ -94,19 +97,23 @@ func NewLSMMetrics(reg prometheus.Registerer) *LSMMetrics {
 }
 
 // NewLSM returns an LSM-like index of len(levels) levels.
-func NewLSM(prefix string, schema *dynparquet.Schema, levels []*LevelConfig, options ...LSMOption) (*LSM, error) {
+// wait is a function that will block until the given transaction has been committed; this is used only during compaction to ensure
+// that all the tx in the level up to the compaction tx have been committed before compacting.
+func NewLSM(prefix string, schema *dynparquet.Schema, levels []*LevelConfig, wait func(uint64), options ...LSMOption) (*LSM, error) {
 	if err := validateLevels(levels); err != nil {
 		return nil, err
 	}
 
 	lsm := &LSM{
-		schema:     schema,
-		prefix:     prefix,
-		levels:     NewList(L0),
-		sizes:      make([]atomic.Int64, len(levels)),
-		configs:    levels,
-		compacting: &atomic.Bool{},
-		logger:     log.NewNopLogger(),
+		schema:        schema,
+		prefix:        prefix,
+		maxTXRecoverd: make([]uint64, len(levels)),
+		levels:        NewList(L0),
+		sizes:         make([]atomic.Int64, len(levels)),
+		configs:       levels,
+		compacting:    &atomic.Bool{},
+		logger:        log.NewNopLogger(),
+		wait:          wait,
 	}
 
 	for _, opt := range options {
@@ -167,7 +174,7 @@ func (l *LSM) MaxLevel() SentinelType {
 func (l *LSM) Add(tx uint64, record arrow.Record) {
 	record.Retain()
 	size := util.TotalRecordSize(record)
-	l.levels.Prepend(parts.NewArrowPart(tx, record, uint64(size), l.schema, parts.WithCompactionLevel(int(L0))))
+	l.levels.Insert(parts.NewArrowPart(tx, record, uint64(size), l.schema, parts.WithCompactionLevel(int(L0))))
 	l0 := l.sizes[L0].Add(int64(size))
 	l.metrics.LevelSize.WithLabelValues(L0.String()).Set(float64(l0))
 	if l0 >= l.configs[L0].MaxSize {
@@ -186,9 +193,29 @@ func (l *LSM) WaitForPendingCompactions() {
 }
 
 // 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)
+// This should only be used during snapshot recovery. It will drop the insert on the floor if the part is older than a part in the next level of the LSM. This indicates
+// that this part is already accounted for in the next level vis compaction.
+func (l *LSM) InsertPart(part parts.Part) {
+	level := SentinelType(part.CompactionLevel())
+	// Check the next levels if there is one to see if this part should be inserted.
+	if level != l.MaxLevel() {
+		for i := level + 1; i < l.MaxLevel()+1; i++ {
+			if part.TX() <= l.maxTXRecoverd[i] {
+				return
+			}
+		}
+	}
+
+	if r := part.Record(); r != nil {
+		r.Retain()
+	}
+
+	if tx := part.TX(); tx > l.maxTXRecoverd[level] {
+		l.maxTXRecoverd[level] = tx
+	}
+
+	// Insert the part into the correct level, but do not do this if parts with newer TXs have already been inserted.
+	l.findLevel(level).Insert(part)
 	size := l.sizes[level].Add(int64(part.Size()))
 	l.metrics.LevelSize.WithLabelValues(level.String()).Set(float64(size))
 }
@@ -334,9 +361,20 @@ func (l *LSM) merge(level SentinelType, externalWriter func([]parts.Part) (parts
 	}
 	l.metrics.Compactions.WithLabelValues(level.String()).Inc()
 
+	compact := l.findLevel(level)
+
+	// Wait for the watermark to reach the most recent transaction.
+	// This ensures a sorted list of transactions.
+	if level == L0 {
+		compact = compact.next.Load()
+		if compact == nil || compact.part == nil {
+			return nil // nothing to compact
+		}
+		l.wait(compact.part.TX())
+	}
+
 	nodeList := []*Node{}
 	var next *Node
-	compact := l.findLevel(level)
 	var iterErr error
 	compact.Iterate(func(node *Node) bool {
 		if node.part == nil { // sentinel encountered
diff --git a/index/lsm_list.go b/index/lsm_list.go
index 50c2d61..1c68bcb 100644
--- a/index/lsm_list.go
+++ b/index/lsm_list.go
@@ -2,6 +2,7 @@ package index
 
 import (
 	"fmt"
+	"runtime"
 	"sync/atomic"
 
 	"github.com/polarsignals/frostdb/parts"
@@ -75,6 +76,32 @@ func (n *Node) Prepend(part parts.Part) *Node {
 	})
 }
 
+// Insert a Node into the list, in order by Tx.
+func (n *Node) Insert(part parts.Part) {
+	node := &Node{
+		part: part,
+	}
+	tx := node.part.TX()
+	tryInsert := func() bool {
+		prev := n
+		next := n.next.Load()
+		for {
+			if next == nil {
+				return prev.next.CompareAndSwap(next, node)
+			}
+			if next.part == nil || next.part.TX() < tx {
+				node.next.Store(next)
+				return prev.next.CompareAndSwap(next, node)
+			}
+			prev = next
+			next = next.next.Load()
+		}
+	}
+	for !tryInsert() {
+		runtime.Gosched()
+	}
+}
+
 func (n *Node) prepend(node *Node) *Node {
 	for { // continue until a successful compare and swap occurs
 		next := n.next.Load()
diff --git a/snapshot.go b/snapshot.go
index 7be6270..f118b7f 100644
--- a/snapshot.go
+++ b/snapshot.go
@@ -68,6 +68,8 @@ const (
 	minReadVersion = snapshotVersion
 )
 
+var ErrSkipPart = fmt.Errorf("skip part")
+
 type snapshotMetrics struct {
 	snapshotsTotal            *prometheus.CounterVec
 	snapshotFileSizeBytes     prometheus.Gauge
@@ -438,7 +440,7 @@ func WriteSnapshot(ctx context.Context, _ uint64, db *DB, w io.Writer, offline b
 					Tx:              p.TX(),
 					CompactionLevel: uint64(p.CompactionLevel()),
 				}
-				if err := func() error {
+				err := func() error {
 					if err := ctx.Err(); err != nil {
 						return err
 					}
@@ -453,6 +455,14 @@ func WriteSnapshot(ctx context.Context, _ uint64, db *DB, w io.Writer, offline b
 						defer recordWriter.Close()
 						return recordWriter.Write(record)
 					}
+
+					// If file compaction is enabled we do not need to snapshot the parquet parts.
+					for _, cfg := range db.columnStore.indexConfig {
+						if cfg.Type == CompactionTypeParquetDisk {
+							return ErrSkipPart
+						}
+					}
+
 					partMeta.Encoding = snapshotpb.Part_ENCODING_PARQUET
 
 					buf, err := p.AsSerializedBuffer(schema)
@@ -467,10 +477,15 @@ func WriteSnapshot(ctx context.Context, _ uint64, db *DB, w io.Writer, offline b
 						return err
 					}
 					return nil
-				}(); err != nil {
+				}()
+				if err != nil {
+					if err == ErrSkipPart {
+						return true
+					}
 					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
@@ -588,20 +603,20 @@ func loadSnapshot(ctx context.Context, db *DB, r io.ReaderAt, size int64) error
 				schemaMsg,
 				options...,
 			)
-			table, err := db.Table(tableMeta.Name, tableConfig)
-			if err != nil {
-				return err
-			}
 
 			var blockUlid ulid.ULID
 			if err := blockUlid.UnmarshalBinary(tableMeta.ActiveBlock.Ulid); err != nil {
 				return err
 			}
 
+			table, err := db.table(tableMeta.Name, tableConfig, blockUlid)
+			if err != nil {
+				return err
+			}
+
 			table.mtx.Lock()
 			block := table.active
 			block.mtx.Lock()
-			block.ulid = blockUlid
 			// Store the last snapshot size so a snapshot is not triggered right
 			// after loading this snapshot.
 			block.lastSnapshotSize.Store(tableMeta.ActiveBlock.Size)
@@ -659,7 +674,7 @@ func loadSnapshot(ctx context.Context, db *DB, r io.ReaderAt, size int64) error
 				}
 
 				for _, part := range resultParts {
-					newIdx.InsertPart(index.SentinelType(part.CompactionLevel()), part)
+					newIdx.InsertPart(part)
 				}
 			}
 
diff --git a/table.go b/table.go
index 80c00ab..6a69085 100644
--- a/table.go
+++ b/table.go
@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
-	"os"
 	"path/filepath"
 	"runtime"
 	"sort"
@@ -276,7 +275,10 @@ type Table struct {
 
 	wal     WAL
 	closing bool
-	closers []io.Closer
+}
+
+type Sync interface {
+	Sync() error
 }
 
 type WAL interface {
@@ -316,7 +318,9 @@ type TableBlock struct {
 	// snapshot was last triggered.
 	lastSnapshotSize atomic.Int64
 
-	index *index.LSM
+	index   *index.LSM
+	closers []Closer
+	syncers []Sync
 
 	pendingWritersWg sync.WaitGroup
 	pendingReadersWg sync.WaitGroup
@@ -324,6 +328,10 @@ type TableBlock struct {
 	mtx *sync.RWMutex
 }
 
+type Closer interface {
+	Close(cleanup bool) error
+}
+
 type tableMetrics struct {
 	blockPersisted       prometheus.Counter
 	blockRotated         prometheus.Counter
@@ -618,7 +626,12 @@ func (t *Table) RotateBlock(_ context.Context, block *TableBlock, skipPersist bo
 	// We don't check t.db.columnStore.manualBlockRotation here because this is
 	// the entry point for users to trigger a manual block rotation and they
 	// will specify through skipPersist if they want the block to be persisted.
-	go t.writeBlock(block, skipPersist, true)
+	go func() {
+		t.writeBlock(block, skipPersist, true)
+		if err := block.Close(true); err != nil {
+			level.Error(t.logger).Log("msg", "closing block", "err", err)
+		}
+	}()
 
 	return nil
 }
@@ -950,15 +963,25 @@ func newTableBlock(table *Table, prevTx, tx uint64, id ulid.ULID) (*TableBlock,
 		prevTx: prevTx,
 	}
 
-	var err error
+	cfg, recovered, err := tb.configureLSMLevels(table.db.columnStore.indexConfig)
+	if err != nil {
+		return nil, fmt.Errorf("configure LSM levels: %w", err)
+	}
+
 	tb.index, err = index.NewLSM(
 		table.name,
 		table.schema,
-		table.configureLSMLevels(table.db.columnStore.indexConfig),
+		cfg,
+		table.db.Wait,
 		index.LSMWithMetrics(table.metrics.indexMetrics),
 	)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("new LSM: %w", err)
+	}
+
+	// Inserts parts into the index that were recovered.
+	for _, p := range recovered {
+		tb.index.InsertPart(p)
 	}
 
 	return tb, nil
@@ -1006,6 +1029,16 @@ func (t *TableBlock) Serialize(writer io.Writer) error {
 	return t.index.Rotate(t.index.MaxLevel(), t.table.externalParquetCompaction(writer))
 }
 
+// Close the block and release all resources.
+func (t *TableBlock) Close(cleanup bool) error {
+	for _, closer := range t.closers {
+		if err := closer.Close(cleanup); err != nil {
+			return fmt.Errorf("closer: %w", err)
+		}
+	}
+	return nil
+}
+
 type ParquetWriter interface {
 	Flush() error
 	WriteRows([]parquet.Row) (int, error)
@@ -1182,10 +1215,8 @@ func (t *Table) close() {
 	t.closing = true
 	t.active.index.WaitForPendingCompactions()
 
-	for _, closer := range t.closers {
-		if err := closer.Close(); err != nil {
-			level.Error(t.logger).Log("msg", "table closer", "err", err)
-		}
+	if err := t.active.Close(false); err != nil {
+		level.Error(t.logger).Log("msg", "closing table block", "err", err)
 	}
 }
 
@@ -1203,31 +1234,52 @@ type IndexConfig struct {
 	Type    CompactionType
 }
 
-// configureLSMLevels configures the level configs for this table.
-func (t *Table) configureLSMLevels(levels []*IndexConfig) []*index.LevelConfig {
-	config := make([]*index.LevelConfig, 0, len(levels))
+// configureLSMLevels configures the level configs for this table block.
+func (t *TableBlock) configureLSMLevels(levels []*IndexConfig) ([]*index.LevelConfig, []parts.Part, error) {
+	config := make([]*index.LevelConfig, len(levels))
+	recovered := []parts.Part{}
+	abortRecovery := false
 
-	for i, level := range levels {
+	// Recover in reverse order so that the highest level is recovered first.
+	// This allows us to throw away parts that were compacted into a higher level but for some reason weren't successfully removed.
+	for i := len(levels) - 1; i >= 0; i-- {
+		lvl := levels[i]
 		cfg := &index.LevelConfig{
-			Level:   index.SentinelType(level.Level),
-			MaxSize: level.MaxSize,
+			Level:   index.SentinelType(lvl.Level),
+			MaxSize: lvl.MaxSize,
 		}
-		switch level.Type {
+		switch lvl.Type {
 		case CompactionTypeParquet:
-			cfg.Compact = t.parquetCompaction
+			cfg.Compact = t.table.parquetCompaction
 		case CompactionTypeParquetDisk:
-			fileCompaction := t.parquetFileCompaction(i + 1)
+			fileCompaction := NewFileCompaction(t.table, t.ulid, i+1)
+			if abortRecovery { // If we failed to recover an LSM file, we need to abort recovery of all lower levels as well, and let the WAL recover.
+				if err := fileCompaction.Truncate(); err != nil {
+					return nil, nil, err
+				}
+			} else {
+				parts, err := fileCompaction.recover(parts.WithCompactionLevel(i + 1))
+				if err != nil {
+					if errors.Is(ErrCompactionRecoveryFailed, err) {
+						abortRecovery = true
+					} else {
+						return nil, nil, err
+					}
+				}
+				recovered = append(recovered, parts...)
+			}
 			t.closers = append(t.closers, fileCompaction) // Append to closers so that the underlying files are closed on table close.
+			t.syncers = append(t.syncers, fileCompaction) // Append to syncers so that the underlying files are synced on wal truncation.
 			cfg.Compact = fileCompaction.writeRecordsToParquetFile
 		default:
 			if i != len(levels)-1 { // Compaction type should not be set for last level
-				panic(fmt.Sprintf("unknown compaction type: %v", level.Type))
+				panic(fmt.Sprintf("unknown compaction type: %v", lvl.Type))
 			}
 		}
-		config = append(config, cfg)
+		config[i] = cfg
 	}
 
-	return config
+	return config, recovered, nil
 }
 
 func (t *Table) parquetCompaction(compact []parts.Part, options ...parts.Option) ([]parts.Part, int64, int64, error) {
@@ -1276,7 +1328,7 @@ func (t *Table) externalParquetCompaction(writer io.Writer) func(compact []parts
 
 // 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) {
+func (t *Table) compactParts(w io.Writer, compact []parts.Part, options ...parquet.WriterOption) (int64, error) {
 	preCompactionSize := int64(0)
 	for _, p := range compact {
 		preCompactionSize += p.Size()
@@ -1314,12 +1366,21 @@ func (t *Table) compactParts(w io.Writer, compact []parts.Part) (int64, error) {
 		return 0, err
 	}
 	err = func() error {
-		pw, err := t.schema.GetWriter(w, merged.DynamicColumns(), false)
-		if err != nil {
-			return err
+		var writer dynparquet.ParquetWriter
+		if len(options) > 0 {
+			writer, err = t.schema.NewWriter(w, merged.DynamicColumns(), false, options...)
+			if err != nil {
+				return err
+			}
+		} else {
+			pw, err := t.schema.GetWriter(w, merged.DynamicColumns(), false)
+			if err != nil {
+				return err
+			}
+			defer t.schema.PutWriter(pw)
+			writer = pw.ParquetWriter
 		}
-		defer t.schema.PutWriter(pw)
-		p, err := t.active.rowWriter(pw)
+		p, err := t.active.rowWriter(writer)
 		if err != nil {
 			return err
 		}
@@ -1497,86 +1558,3 @@ func (t *Table) distinctRecordsForCompaction(compact []parts.Part) ([]arrow.Reco
 	}
 	return newRecords, nil
 }
-
-type fileCompaction struct {
-	t      *Table
-	file   *os.File
-	offset int64 // Writing offsets into the file
-	ref    int64 // Number of references to file.
-}
-
-func (f *fileCompaction) Close() error {
-	return f.file.Close()
-}
-
-func (t *Table) parquetFileCompaction(lvl int) *fileCompaction {
-	f := &fileCompaction{
-		t: t,
-	}
-
-	file, err := os.CreateTemp("", fmt.Sprintf("L%v-*.parquet", lvl))
-	if err != nil {
-		panic(err)
-	}
-	level.Info(t.logger).Log("msg", "created temp file for level", "level", lvl, "file", file.Name())
-	f.file = file
-
-	return f
-}
-
-// accountingWriter is a writer that accounts for the number of bytes written.
-type accountingWriter struct {
-	w io.Writer
-	n int64
-}
-
-func (a *accountingWriter) Write(p []byte) (int, error) {
-	n, err := a.w.Write(p)
-	a.n += int64(n)
-	return n, err
-}
-
-// writeRecordsToParquetFile will compact the given parts into a Parquet file written to the next level file.
-func (f *fileCompaction) writeRecordsToParquetFile(compact []parts.Part, options ...parts.Option) ([]parts.Part, int64, int64, error) {
-	accountant := &accountingWriter{w: f.file}
-	preCompactionSize, err := f.t.compactParts(accountant, compact) // compact into the next level
-	if err != nil {
-		return nil, 0, 0, err
-	}
-
-	// Record the writing offset into the file.
-	prevOffset := f.offset
-	f.offset += accountant.n
-	f.ref++
-
-	pf, err := parquet.OpenFile(io.NewSectionReader(f.file, prevOffset, accountant.n), accountant.n)
-	if err != nil {
-		return nil, 0, 0, err
-	}
-
-	buf, err := dynparquet.NewSerializedBuffer(pf)
-	if err != nil {
-		return nil, 0, 0, err
-	}
-
-	return []parts.Part{parts.NewParquetPart(0, buf, append(options, parts.WithRelease(f.release()))...)}, preCompactionSize, accountant.n, nil
-}
-
-// release will account for all the Parts currently pointing to this file. Once the last one has been released it will truncate the file.
-// release is not safe to call concurrently.
-func (f *fileCompaction) release() func() {
-	return func() {
-		f.ref--
-		if f.ref == 0 {
-			level.Info(f.t.logger).Log("msg", "truncating file", "file", f.file.Name())
-			if err := os.Truncate(f.file.Name(), 0); err != nil {
-				panic(fmt.Errorf("failed to truncate the level file: %v", err))
-			}
-
-			f.offset = 0
-			if _, err := f.file.Seek(0, io.SeekStart); err != nil {
-				panic(fmt.Errorf("failed to seek the level file: %v", err))
-			}
-		}
-	}
-}


Test patch:
diff --git a/db_test.go b/db_test.go
index 3dac5bb..81fc1ec 100644
--- a/db_test.go
+++ b/db_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/polarsignals/frostdb/dynparquet"
 	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
 	walpb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/wal/v1alpha1"
+	"github.com/polarsignals/frostdb/index"
 	"github.com/polarsignals/frostdb/query"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 	"github.com/polarsignals/frostdb/query/physicalplan"
@@ -2271,3 +2272,326 @@ func Test_DB_WithParquetDiskCompaction(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(300), rows)
 }
+
+func Test_DB_PersistentDiskCompaction(t *testing.T) {
+	t.Parallel()
+	config := NewTableConfig(
+		dynparquet.SampleDefinition(),
+	)
+	logger := newTestLogger(t)
+
+	tests := map[string]struct {
+		beforeReplay func(table *Table, dir string)
+		beforeClose  func(table *Table, dir string)
+		lvl2Parts    int
+		lvl1Parts    int
+		finalRows    int64
+	}{
+		"happy path": {
+			beforeReplay: func(_ *Table, _ string) {},
+			beforeClose:  func(_ *Table, _ string) {},
+			lvl2Parts:    3,
+			lvl1Parts:    1,
+			finalRows:    1200,
+		},
+		"corrupted L1": {
+			beforeClose: func(table *Table, dir string) {},
+			lvl2Parts:   3,
+			lvl1Parts:   1,
+			finalRows:   1200,
+			beforeReplay: func(table *Table, dir string) {
+				// Corrupt the LSM file; this should trigger a recovery from the WAL
+				levelFile := filepath.Join(dir, "databases", "test", "index", "test", fmt.Sprintf("L1-%s.parquet", table.active.ulid.String()))
+				info, err := os.Stat(levelFile)
+				require.NoError(t, err)
+				require.NoError(t, os.Truncate(levelFile, info.Size()-1)) // truncate the last byte to pretend the write didn't finish
+			},
+		},
+		"corrupted L2": {
+			beforeClose: func(table *Table, dir string) {},
+			lvl2Parts:   3,
+			lvl1Parts:   1,
+			finalRows:   1200,
+			beforeReplay: func(table *Table, dir string) {
+				// Corrupt the LSM file; this should trigger a recovery from the WAL
+				levelFile := filepath.Join(dir, "databases", "test", "index", "test", fmt.Sprintf("L2-%s.parquet", table.active.ulid.String()))
+				info, err := os.Stat(levelFile)
+				require.NoError(t, err)
+				require.NoError(t, os.Truncate(levelFile, info.Size()-1)) // truncate the last byte to pretend the write didn't finish
+			},
+		},
+		"untruncated L1": { // Test that if we fail to truncate the L1 file, we can still recover
+			lvl2Parts:    4,
+			lvl1Parts:    1,
+			finalRows:    1200,
+			beforeReplay: func(_ *Table, _ string) {},
+			beforeClose: func(table *Table, dir string) { // trigger a compaction, but save off the L1 file and move it back. This mimics a failure to truncate the L1 file
+				// Save the L1 file
+				levelFile := filepath.Join(dir, "databases", "test", "index", "test", fmt.Sprintf("L1-%s.parquet", table.active.ulid.String()))
+				src, err := os.Open(levelFile)
+				require.NoError(t, err)
+
+				saveFile := filepath.Join(dir, "databases", "test", "index", "test", fmt.Sprintf("L1-%s.parquet.sav", table.active.ulid.String()))
+				dst, err := os.Create(saveFile)
+				require.NoError(t, err)
+
+				_, err = io.Copy(dst, src)
+				require.NoError(t, err)
+				dst.Close()
+				src.Close()
+
+				// This truncates the L1 file
+				require.NoError(t, table.EnsureCompaction())
+
+				// Move the L1 file back
+				require.NoError(t, os.Rename(saveFile, levelFile))
+			},
+		},
+		"snapshot": {
+			beforeReplay: func(_ *Table, _ string) {},
+			beforeClose: func(table *Table, dir string) {
+				// trigger a snapshot
+				tx := table.db.highWatermark.Load()
+				require.NoError(t, table.db.snapshotAtTX(context.Background(), tx, table.db.snapshotWriter(tx)))
+
+				// Write more to the table and trigger a compaction
+				samples := dynparquet.NewTestSamples()
+				for j := 0; j < 100; j++ {
+					r, err := samples.ToRecord()
+					require.NoError(t, err)
+					_, err = table.InsertRecord(context.Background(), r)
+					require.NoError(t, err)
+				}
+				require.Eventually(t, func() bool {
+					return table.active.index.LevelSize(index.L1) != 0
+				}, time.Second, time.Millisecond*10)
+			},
+			lvl2Parts: 3,
+			lvl1Parts: 2,
+			finalRows: 1500,
+		},
+	}
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			dir := t.TempDir()
+			cfg := []*IndexConfig{
+				{Level: int(index.L0), MaxSize: 25700, Type: CompactionTypeParquetDisk},
+				{Level: int(index.L1), MaxSize: 1024 * 1024 * 128, Type: CompactionTypeParquetDisk},
+				{Level: int(index.L2), MaxSize: 1024 * 1024 * 512},
+			}
+			c, err := New(
+				WithLogger(logger),
+				WithIndexConfig(cfg),
+				WithStoragePath(dir),
+				WithWAL(),
+			)
+			t.Cleanup(func() {
+				require.NoError(t, c.Close())
+			})
+			require.NoError(t, err)
+			db, err := c.DB(context.Background(), "test")
+			require.NoError(t, err)
+			table, err := db.Table("test", config)
+			require.NoError(t, err)
+
+			samples := dynparquet.NewTestSamples()
+
+			ctx := context.Background()
+			compactions := 3 // 3 compacted parts in L2 file
+			for i := 0; i < compactions; i++ {
+				for j := 0; j < 100; j++ {
+					r, err := samples.ToRecord()
+					require.NoError(t, err)
+					_, err = table.InsertRecord(ctx, r)
+					require.NoError(t, err)
+				}
+				require.NoError(t, table.EnsureCompaction())
+			}
+
+			for j := 0; j < 100; j++ {
+				r, err := samples.ToRecord()
+				require.NoError(t, err)
+				_, err = table.InsertRecord(ctx, r)
+				require.NoError(t, err)
+			}
+			require.Eventually(t, func() bool {
+				return table.active.index.LevelSize(index.L1) != 0
+			}, time.Second, time.Millisecond*10)
+
+			// Ensure that disk compacted data can be recovered
+			pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+			defer pool.AssertSize(t, 0)
+			rows := int64(0)
+			engine := query.NewEngine(pool, db.TableProvider())
+			err = engine.ScanTable("test").
+				Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
+					rows += r.NumRows()
+					return nil
+				})
+			require.NoError(t, err)
+			require.Equal(t, int64((300*compactions)+300), rows)
+
+			test.beforeClose(table, dir)
+			id := table.active.ulid
+
+			// Close the database
+			wm := db.highWatermark.Load()
+			require.NoError(t, c.Close())
+
+			fc := NewFileCompaction(table, id, 2)
+			p, err := fc.recover()
+			require.NoError(t, err)
+			require.Equal(t, test.lvl2Parts, len(p))
+
+			for i := 0; i < compactions; i++ {
+				buf, err := p[i].AsSerializedBuffer(nil)
+				require.NoError(t, err)
+				tx, ok := buf.ParquetFile().Lookup("compaction_tx")
+				require.True(t, ok)
+				require.Equal(t, fmt.Sprintf("%v", test.lvl2Parts*100-(i*100)+1), tx)
+			}
+
+			fc = NewFileCompaction(table, id, 1)
+			p, err = fc.recover()
+			require.NoError(t, err)
+			require.Equal(t, test.lvl1Parts, len(p))
+
+			buf, err := p[0].AsSerializedBuffer(nil)
+			require.NoError(t, err)
+			tx, ok := buf.ParquetFile().Lookup("compaction_tx")
+			require.True(t, ok)
+			require.Equal(t, fmt.Sprintf("%v", wm), tx)
+
+			// Run the beforeReplay hook
+			test.beforeReplay(table, dir)
+
+			// Reopen database; expect it to recover from the LSM files and WAL
+			c, err = New(
+				WithLogger(logger),
+				WithIndexConfig(cfg),
+				WithStoragePath(dir),
+				WithWAL(),
+			)
+			require.NoError(t, err)
+			t.Cleanup(func() {
+				require.NoError(t, c.Close())
+			})
+			db, err = c.DB(context.Background(), "test")
+			require.NoError(t, err)
+
+			rows = int64(0)
+			engine = query.NewEngine(pool, db.TableProvider())
+			err = engine.ScanTable("test").
+				Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
+					rows += r.NumRows()
+					return nil
+				})
+			require.NoError(t, err)
+			require.Equal(t, test.finalRows, rows)
+		})
+	}
+}
+
+// Ensure data integrity when a block is rotated.
+func Test_DB_PersistentDiskCompaction_BlockRotation(t *testing.T) {
+	t.Parallel()
+	config := NewTableConfig(
+		dynparquet.SampleDefinition(),
+	)
+	logger := newTestLogger(t)
+	bucket := objstore.NewInMemBucket()
+	sinksource := NewDefaultObjstoreBucket(bucket)
+
+	dir := t.TempDir()
+	cfg := []*IndexConfig{
+		{Level: int(index.L0), MaxSize: 25700, Type: CompactionTypeParquetDisk},
+		{Level: int(index.L1), MaxSize: 1024 * 1024 * 128, Type: CompactionTypeParquetDisk},
+		{Level: int(index.L2), MaxSize: 1024 * 1024 * 512},
+	}
+	c, err := New(
+		WithLogger(logger),
+		WithIndexConfig(cfg),
+		WithStoragePath(dir),
+		WithWAL(),
+		WithManualBlockRotation(),
+		WithReadWriteStorage(sinksource),
+	)
+	t.Cleanup(func() {
+		require.NoError(t, c.Close())
+	})
+	require.NoError(t, err)
+	db, err := c.DB(context.Background(), "test")
+	require.NoError(t, err)
+	table, err := db.Table("test", config)
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+
+	ctx := context.Background()
+	compactions := 3 // 3 compacted parts in L2 file
+	for i := 0; i < compactions; i++ {
+		for j := 0; j < 100; j++ {
+			r, err := samples.ToRecord()
+			require.NoError(t, err)
+			_, err = table.InsertRecord(ctx, r)
+			require.NoError(t, err)
+		}
+		require.NoError(t, table.EnsureCompaction())
+	}
+
+	// Get data in L1
+	for j := 0; j < 100; j++ {
+		r, err := samples.ToRecord()
+		require.NoError(t, err)
+		_, err = table.InsertRecord(ctx, r)
+		require.NoError(t, err)
+	}
+	require.Eventually(t, func() bool {
+		return table.active.index.LevelSize(index.L1) != 0
+	}, time.Second, time.Millisecond*10)
+
+	validateRows := func(expected int64) {
+		pool := memory.NewCheckedAllocator(memory.DefaultAllocator)
+		defer pool.AssertSize(t, 0)
+		rows := int64(0)
+		engine := query.NewEngine(pool, db.TableProvider())
+		err = engine.ScanTable("test").
+			Execute(context.Background(), func(ctx context.Context, r arrow.Record) error {
+				rows += r.NumRows()
+				return nil
+			})
+		require.NoError(t, err)
+		require.Equal(t, expected, rows)
+	}
+
+	validateRows(1200)
+
+	// Rotate block
+	require.NoError(t, table.RotateBlock(context.Background(), table.ActiveBlock(), false))
+
+	validateRows(1200)
+
+	for i := 0; i < compactions; i++ {
+		for j := 0; j < 100; j++ {
+			r, err := samples.ToRecord()
+			require.NoError(t, err)
+			_, err = table.InsertRecord(ctx, r)
+			require.NoError(t, err)
+		}
+		require.NoError(t, table.EnsureCompaction())
+	}
+
+	// Get data in L1
+	for j := 0; j < 100; j++ {
+		r, err := samples.ToRecord()
+		require.NoError(t, err)
+		_, err = table.InsertRecord(ctx, r)
+		require.NoError(t, err)
+	}
+	require.Eventually(t, func() bool {
+		return table.active.index.LevelSize(index.L1) != 0
+	}, time.Second, time.Millisecond*10)
+
+	validateRows(2400)
+}
diff --git a/index/lsm_test.go b/index/lsm_test.go
index 92d0970..fc47426 100644
--- a/index/lsm_test.go
+++ b/index/lsm_test.go
@@ -5,6 +5,10 @@ import (
 	"context"
 	"errors"
 	"io"
+	"math"
+	"math/rand"
+	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -87,7 +91,7 @@ func check(t *testing.T, lsm *LSM, records, buffers int) {
 	})
 	rec := 0
 	buf := 0
-	require.NoError(t, lsm.Scan(context.Background(), "", nil, nil, 2, func(ctx context.Context, v any) error {
+	require.NoError(t, lsm.Scan(context.Background(), "", nil, nil, math.MaxUint64, func(ctx context.Context, v any) error {
 		switch v.(type) {
 		case arrow.Record:
 			rec++
@@ -106,7 +110,9 @@ func Test_LSM_Basic(t *testing.T) {
 		{Level: L0, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
 		{Level: L1, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
 		{Level: L2, MaxSize: 1024 * 1024 * 1024},
-	})
+	},
+		func(uint64) {},
+	)
 	require.NoError(t, err)
 
 	samples := dynparquet.NewTestSamples()
@@ -114,18 +120,18 @@ func Test_LSM_Basic(t *testing.T) {
 	require.NoError(t, err)
 
 	lsm.Add(1, r)
-	lsm.Add(1, r)
-	lsm.Add(1, r)
+	lsm.Add(2, r)
+	lsm.Add(3, r)
 	check(t, lsm, 3, 0)
 	require.NoError(t, lsm.merge(L0, nil))
 	check(t, lsm, 0, 1)
-	lsm.Add(1, r)
+	lsm.Add(4, r)
 	check(t, lsm, 1, 1)
-	lsm.Add(1, r)
+	lsm.Add(5, r)
 	check(t, lsm, 2, 1)
 	require.NoError(t, lsm.merge(L0, nil))
 	check(t, lsm, 0, 2)
-	lsm.Add(1, r)
+	lsm.Add(6, r)
 	check(t, lsm, 1, 2)
 	require.NoError(t, lsm.merge(L1, nil))
 	check(t, lsm, 1, 1)
@@ -139,7 +145,9 @@ func Test_LSM_DuplicateSentinel(t *testing.T) {
 		{Level: L0, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
 		{Level: L1, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
 		{Level: L2, MaxSize: 1024 * 1024 * 1024},
-	})
+	},
+		func(uint64) {},
+	)
 	require.NoError(t, err)
 
 	samples := dynparquet.NewTestSamples()
@@ -147,8 +155,8 @@ func Test_LSM_DuplicateSentinel(t *testing.T) {
 	require.NoError(t, err)
 
 	lsm.Add(1, r)
-	lsm.Add(1, r)
-	lsm.Add(1, r)
+	lsm.Add(2, r)
+	lsm.Add(3, r)
 	check(t, lsm, 3, 0)
 	require.NoError(t, lsm.merge(L0, nil))
 	check(t, lsm, 0, 1)
@@ -161,7 +169,9 @@ func Test_LSM_Compaction(t *testing.T) {
 	lsm, err := NewLSM("test", nil, []*LevelConfig{
 		{Level: L0, MaxSize: 1, Compact: parquetCompaction},
 		{Level: L1, MaxSize: 1024 * 1024 * 1024},
-	})
+	},
+		func(uint64) {},
+	)
 	require.NoError(t, err)
 
 	samples := dynparquet.NewTestSamples()
@@ -171,7 +181,7 @@ func Test_LSM_Compaction(t *testing.T) {
 	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)
+	}, 3*time.Second, 10*time.Millisecond)
 }
 
 func Test_LSM_CascadeCompaction(t *testing.T) {
@@ -182,7 +192,9 @@ func Test_LSM_CascadeCompaction(t *testing.T) {
 		{Level: L2, MaxSize: 2281, Compact: parquetCompaction},
 		{Level: 3, MaxSize: 2281, Compact: parquetCompaction},
 		{Level: 4, MaxSize: 2281},
-	})
+	},
+		func(uint64) {},
+	)
 	require.NoError(t, err)
 
 	samples := dynparquet.NewTestSamples()
@@ -196,13 +208,62 @@ func Test_LSM_CascadeCompaction(t *testing.T) {
 			lsm.sizes[L2].Load() == 0 &&
 			lsm.sizes[3].Load() == 0 &&
 			lsm.sizes[4].Load() == 0
-	}, time.Second, 5*time.Millisecond)
-	lsm.Add(1, r)
+	}, 3*time.Second, 10*time.Millisecond)
+	lsm.Add(2, 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)
+	}, 3*time.Second, 10*time.Millisecond)
+}
+
+func Test_LSM_InOrderInsert(t *testing.T) {
+	t.Parallel()
+	lsm, err := NewLSM("test", nil, []*LevelConfig{
+		{Level: L0, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L1, MaxSize: 1024 * 1024 * 1024, Compact: parquetCompaction},
+		{Level: L2, MaxSize: 1024 * 1024 * 1024},
+	},
+		func(uint64) {},
+	)
+	require.NoError(t, err)
+
+	samples := dynparquet.NewTestSamples()
+	r, err := samples.ToRecord()
+	require.NoError(t, err)
+
+	wg := &sync.WaitGroup{}
+	workers := 100
+	inserts := 100
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func() {
+			defer wg.Done()
+			for j := 0; j < inserts; j++ {
+				lsm.Add(rand.Uint64(), r)
+			}
+		}()
+	}
+	wg.Wait()
+
+	tx := make([]uint64, 0, workers*inserts)
+	lsm.Iterate(func(node *Node) bool {
+		if node.part != nil {
+			tx = append(tx, node.part.TX())
+		}
+		return true
+	})
+
+	// check that the transactions are sorted in descending order
+	require.True(t, slices.IsSortedFunc[[]uint64, uint64](tx, func(i, j uint64) int {
+		if i < j {
+			return 1
+		} else if i > j {
+			return -1
+		}
+
+		return 0
+	}))
 }