Repository: polarsignals/frostdb
Pull request #633 (2023-12-15 12:30:26)

pqarrow/arrowutils: Add optional limit to MergeRecords
This is helpful for merging multiple sorted records but only keeping the first e.g. 1000 rows.


metalmatze: Finally resolved the small mistake and this should be ready for another review with `uint64` now.

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..6b280d1 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"
 )
@@ -51,7 +53,6 @@ type ColumnStore struct {
 	reg                 prometheus.Registerer
 	logger              log.Logger
 	tracer              trace.Tracer
-	granuleSizeBytes    int64
 	activeMemorySize    int64
 	storagePath         string
 	enableWAL           bool
@@ -99,7 +100,6 @@ func New(
 		indexConfig:         DefaultIndexConfig(),
 		indexDegree:         2,
 		splitSize:           2,
-		granuleSizeBytes:    1 * MiB,
 		activeMemorySize:    512 * MiB,
 	}
 
@@ -128,6 +128,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
 	}
@@ -156,13 +164,6 @@ func WithRegistry(reg prometheus.Registerer) Option {
 	}
 }
 
-func WithGranuleSizeBytes(bytes int64) Option {
-	return func(s *ColumnStore) error {
-		s.granuleSizeBytes = bytes
-		return nil
-	}
-}
-
 func WithActiveMemorySize(size int64) Option {
 	return func(s *ColumnStore) error {
 		s.activeMemorySize = size
@@ -481,12 +482,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 +673,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 +740,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 +775,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 +898,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 +934,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 +1023,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 +1136,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 +1190,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 +1304,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/pqarrow/arrowutils/merge.go b/pqarrow/arrowutils/merge.go
index e9d9d11..8df2a9c 100644
--- a/pqarrow/arrowutils/merge.go
+++ b/pqarrow/arrowutils/merge.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"container/heap"
 	"fmt"
+	"math"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
@@ -14,11 +15,15 @@ import (
 
 // MergeRecords merges the given records. The records must all have the same
 // schema. orderByCols is a slice of indexes into the columns that the records
-// and resulting records are ordered by. Note that the given records should
-// already be ordered by the given columns.
+// and resulting records are ordered by. While ordering the limit is checked before appending more rows.
+// If limit is less than 1, no limit is applied.
+// Note that the given records should already be ordered by the given columns.
 // WARNING: Only ascending ordering is currently supported.
 func MergeRecords(
-	mem memory.Allocator, records []arrow.Record, orderByCols []int,
+	mem memory.Allocator,
+	records []arrow.Record,
+	orderByCols []int,
+	limit uint64,
 ) (arrow.Record, error) {
 	h := cursorHeap{
 		cursors:     make([]cursor, len(records)),
@@ -32,8 +37,13 @@ func MergeRecords(
 	recordBuilder := builder.NewRecordBuilder(mem, schema)
 	defer recordBuilder.Release()
 
+	if limit == 0 {
+		limit = math.MaxInt64
+	}
+	count := uint64(0)
+
 	heap.Init(&h)
-	for h.Len() > 0 {
+	for h.Len() > 0 && count < limit {
 		// Minimum cursor is always at index 0.
 		r := h.cursors[0].r
 		i := h.cursors[0].curIdx
@@ -45,10 +55,12 @@ func MergeRecords(
 		if int64(i+1) >= r.NumRows() {
 			// Pop the cursor since it has no more data.
 			_ = heap.Pop(&h)
+			count++
 			continue
 		}
 		h.cursors[0].curIdx++
 		heap.Fix(&h, 0)
+		count++
 	}
 
 	return recordBuilder.NewRecord(), nil
diff --git a/query/logicalplan/expr.go b/query/logicalplan/expr.go
index 45e53ff..8f73391 100644
--- a/query/logicalplan/expr.go
+++ b/query/logicalplan/expr.go
@@ -69,6 +69,37 @@ func (o Op) String() string {
 	}
 }
 
+func (o Op) ArrowString() string {
+	switch o {
+	case OpEq:
+		return "equal"
+	case OpNotEq:
+		return "not_equal"
+	case OpLt:
+		return "less"
+	case OpLtEq:
+		return "less_equal"
+	case OpGt:
+		return "greater"
+	case OpGtEq:
+		return "greater_equal"
+	case OpAnd:
+		return "and"
+	case OpOr:
+		return "or"
+	case OpAdd:
+		return "add"
+	case OpSub:
+		return "subtract"
+	case OpMul:
+		return "multiply"
+	case OpDiv:
+		return "divide"
+	default:
+		panic("unknown operator")
+	}
+}
+
 type BinaryExpr struct {
 	Left  Expr
 	Op    Op
diff --git a/query/physicalplan/binaryscalarexpr.go b/query/physicalplan/binaryscalarexpr.go
index 768a520..710156f 100644
--- a/query/physicalplan/binaryscalarexpr.go
+++ b/query/physicalplan/binaryscalarexpr.go
@@ -2,11 +2,13 @@ package physicalplan
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
+	"github.com/apache/arrow/go/v14/arrow/compute"
 	"github.com/apache/arrow/go/v14/arrow/scalar"
 
 	"github.com/polarsignals/frostdb/query/logicalplan"
@@ -79,111 +81,8 @@ func (e BinaryScalarExpr) String() string {
 var ErrUnsupportedBinaryOperation = errors.New("unsupported binary operation")
 
 func BinaryScalarOperation(left arrow.Array, right scalar.Scalar, operator logicalplan.Op) (*Bitmap, error) {
+	// TODO: Figure out dictionary arrays and lists with compute next
 	leftType := left.DataType()
-	switch leftType {
-	case arrow.FixedWidthTypes.Boolean:
-		switch operator {
-		case logicalplan.OpEq:
-			return BooleanArrayScalarEqual(left.(*array.Boolean), right.(*scalar.Boolean))
-		case logicalplan.OpNotEq:
-			return BooleanArrayScalarNotEqual(left.(*array.Boolean), right.(*scalar.Boolean))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case &arrow.FixedSizeBinaryType{ByteWidth: 16}:
-		switch operator {
-		case logicalplan.OpEq:
-			return FixedSizeBinaryArrayScalarEqual(left.(*array.FixedSizeBinary), right.(*scalar.FixedSizeBinary))
-		case logicalplan.OpNotEq:
-			return FixedSizeBinaryArrayScalarNotEqual(left.(*array.FixedSizeBinary), right.(*scalar.FixedSizeBinary))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case arrow.BinaryTypes.String:
-		switch operator {
-		case logicalplan.OpEq:
-			return StringArrayScalarEqual(left.(*array.String), right.(*scalar.String))
-		case logicalplan.OpNotEq:
-			return StringArrayScalarNotEqual(left.(*array.String), right.(*scalar.String))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case arrow.BinaryTypes.Binary:
-		switch operator {
-		case logicalplan.OpEq:
-			switch r := right.(type) {
-			case *scalar.Binary:
-				return BinaryArrayScalarEqual(left.(*array.Binary), r)
-			case *scalar.String:
-				return BinaryArrayScalarEqual(left.(*array.Binary), r.Binary)
-			default:
-				panic("something terrible has happened, this should have errored previously during validation")
-			}
-		case logicalplan.OpNotEq:
-			switch r := right.(type) {
-			case *scalar.Binary:
-				return BinaryArrayScalarNotEqual(left.(*array.Binary), r)
-			case *scalar.String:
-				return BinaryArrayScalarNotEqual(left.(*array.Binary), r.Binary)
-			default:
-				panic("something terrible has happened, this should have errored previously during validation")
-			}
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case arrow.PrimitiveTypes.Int64:
-		switch operator {
-		case logicalplan.OpEq:
-			return Int64ArrayScalarEqual(left.(*array.Int64), right.(*scalar.Int64))
-		case logicalplan.OpNotEq:
-			return Int64ArrayScalarNotEqual(left.(*array.Int64), right.(*scalar.Int64))
-		case logicalplan.OpLt:
-			return Int64ArrayScalarLessThan(left.(*array.Int64), right.(*scalar.Int64))
-		case logicalplan.OpLtEq:
-			return Int64ArrayScalarLessThanOrEqual(left.(*array.Int64), right.(*scalar.Int64))
-		case logicalplan.OpGt:
-			return Int64ArrayScalarGreaterThan(left.(*array.Int64), right.(*scalar.Int64))
-		case logicalplan.OpGtEq:
-			return Int64ArrayScalarGreaterThanOrEqual(left.(*array.Int64), right.(*scalar.Int64))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case arrow.PrimitiveTypes.Int32:
-		switch operator {
-		case logicalplan.OpEq:
-			return Int32ArrayScalarEqual(left.(*array.Int32), right.(*scalar.Int32))
-		case logicalplan.OpNotEq:
-			return Int32ArrayScalarNotEqual(left.(*array.Int32), right.(*scalar.Int32))
-		case logicalplan.OpLt:
-			return Int32ArrayScalarLessThan(left.(*array.Int32), right.(*scalar.Int32))
-		case logicalplan.OpLtEq:
-			return Int32ArrayScalarLessThanOrEqual(left.(*array.Int32), right.(*scalar.Int32))
-		case logicalplan.OpGt:
-			return Int32ArrayScalarGreaterThan(left.(*array.Int32), right.(*scalar.Int32))
-		case logicalplan.OpGtEq:
-			return Int32ArrayScalarGreaterThanOrEqual(left.(*array.Int32), right.(*scalar.Int32))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	case arrow.PrimitiveTypes.Uint64:
-		switch operator {
-		case logicalplan.OpEq:
-			return Uint64ArrayScalarEqual(left.(*array.Uint64), right.(*scalar.Uint64))
-		case logicalplan.OpNotEq:
-			return Uint64ArrayScalarNotEqual(left.(*array.Uint64), right.(*scalar.Uint64))
-		case logicalplan.OpLt:
-			return Uint64ArrayScalarLessThan(left.(*array.Uint64), right.(*scalar.Uint64))
-		case logicalplan.OpLtEq:
-			return Uint64ArrayScalarLessThanOrEqual(left.(*array.Uint64), right.(*scalar.Uint64))
-		case logicalplan.OpGt:
-			return Uint64ArrayScalarGreaterThan(left.(*array.Uint64), right.(*scalar.Uint64))
-		case logicalplan.OpGtEq:
-			return Uint64ArrayScalarGreaterThanOrEqual(left.(*array.Uint64), right.(*scalar.Uint64))
-		default:
-			panic("something terrible has happened, this should have errored previously during validation")
-		}
-	}
-
 	switch arr := left.(type) {
 	case *array.Dictionary:
 		switch operator {
@@ -201,7 +100,38 @@ func BinaryScalarOperation(left arrow.Array, right scalar.Scalar, operator logic
 		panic("TODO: list comparisons unimplemented")
 	}
 
-	return nil, ErrUnsupportedBinaryOperation
+	return ArrayScalarCompute(operator.ArrowString(), left, right)
+}
+
+func ArrayScalarCompute(funcName string, left arrow.Array, right scalar.Scalar) (*Bitmap, error) {
+	equalsResult, err := compute.CallFunction(context.TODO(), funcName, nil, compute.NewDatum(left), compute.NewDatum(right))
+	if err != nil {
+		if errors.Unwrap(err).Error() == "not implemented" {
+			return nil, ErrUnsupportedBinaryOperation
+		}
+		return nil, fmt.Errorf("error calling equal function: %w", err)
+	}
+	defer equalsResult.Release()
+	equalsDatum, ok := equalsResult.(*compute.ArrayDatum)
+	if !ok {
+		return nil, fmt.Errorf("expected *compute.ArrayDatum, got %T", equalsResult)
+	}
+	equalsArray, ok := equalsDatum.MakeArray().(*array.Boolean)
+	if !ok {
+		return nil, fmt.Errorf("expected *array.Boolean, got %T", equalsDatum.MakeArray())
+	}
+	defer equalsArray.Release()
+
+	res := NewBitmap()
+	for i := 0; i < equalsArray.Len(); i++ {
+		if equalsArray.IsNull(i) {
+			continue
+		}
+		if equalsArray.Value(i) {
+			res.AddInt(i)
+		}
+	}
+	return res, nil
 }
 
 func DictionaryArrayScalarNotEqual(left *array.Dictionary, right scalar.Scalar) (*Bitmap, error) {
@@ -283,393 +213,3 @@ func DictionaryArrayScalarEqual(left *array.Dictionary, right scalar.Scalar) (*B
 
 	return res, nil
 }
-
-func FixedSizeBinaryArrayScalarEqual(left *array.FixedSizeBinary, right *scalar.FixedSizeBinary) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if bytes.Equal(left.Value(i), right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func FixedSizeBinaryArrayScalarNotEqual(left *array.FixedSizeBinary, right *scalar.FixedSizeBinary) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if !bytes.Equal(left.Value(i), right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func StringArrayScalarEqual(left *array.String, right *scalar.String) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) == string(right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func StringArrayScalarNotEqual(left *array.String, right *scalar.String) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if left.Value(i) != string(right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func BinaryArrayScalarEqual(left *array.Binary, right *scalar.Binary) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if bytes.Equal(left.Value(i), right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func BinaryArrayScalarNotEqual(left *array.Binary, right *scalar.Binary) (*Bitmap, error) {
-	res := NewBitmap()
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if !bytes.Equal(left.Value(i), right.Data()) {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarEqual(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) == right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarNotEqual(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if left.Value(i) != right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarLessThan(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) < right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarLessThanOrEqual(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) <= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarGreaterThan(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) > right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int64ArrayScalarGreaterThanOrEqual(left *array.Int64, right *scalar.Int64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) >= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarEqual(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) == right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarNotEqual(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if left.Value(i) != right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarLessThan(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) < right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarLessThanOrEqual(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) <= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarGreaterThan(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) > right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Uint64ArrayScalarGreaterThanOrEqual(left *array.Uint64, right *scalar.Uint64) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) >= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func BooleanArrayScalarEqual(left *array.Boolean, right *scalar.Boolean) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) == right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func BooleanArrayScalarNotEqual(left *array.Boolean, right *scalar.Boolean) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) != right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarEqual(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) == right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarNotEqual(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			res.Add(uint32(i))
-			continue
-		}
-		if left.Value(i) != right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarLessThan(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) < right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarLessThanOrEqual(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) <= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarGreaterThan(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) > right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
-
-func Int32ArrayScalarGreaterThanOrEqual(left *array.Int32, right *scalar.Int32) (*Bitmap, error) {
-	res := NewBitmap()
-
-	for i := 0; i < left.Len(); i++ {
-		if left.IsNull(i) {
-			continue
-		}
-		if left.Value(i) >= right.Value {
-			res.Add(uint32(i))
-		}
-	}
-
-	return res, nil
-}
diff --git a/query/physicalplan/ordered_aggregate.go b/query/physicalplan/ordered_aggregate.go
index ad973ec..5f6f0eb 100644
--- a/query/physicalplan/ordered_aggregate.go
+++ b/query/physicalplan/ordered_aggregate.go
@@ -486,7 +486,7 @@ func (a *OrderedAggregate) Finish(ctx context.Context) error {
 		for i := range orderByCols {
 			orderByCols[i] = i
 		}
-		mergedRecord, err := arrowutils.MergeRecords(a.pool, records, orderByCols)
+		mergedRecord, err := arrowutils.MergeRecords(a.pool, records, orderByCols, 0)
 		if err != nil {
 			return err
 		}
diff --git a/query/physicalplan/ordered_synchronizer.go b/query/physicalplan/ordered_synchronizer.go
index ccf7abe..444b630 100644
--- a/query/physicalplan/ordered_synchronizer.go
+++ b/query/physicalplan/ordered_synchronizer.go
@@ -117,7 +117,7 @@ func (o *OrderedSynchronizer) mergeRecordsLocked() (arrow.Record, error) {
 	if err := o.ensureSameSchema(o.sync.data); err != nil {
 		return nil, err
 	}
-	mergedRecord, err := arrowutils.MergeRecords(o.pool, o.sync.data, o.orderByCols)
+	mergedRecord, err := arrowutils.MergeRecords(o.pool, o.sync.data, o.orderByCols, 0)
 	if err != nil {
 		return nil, err
 	}
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..378b083 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)
+	}, 30*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)
+	}, 30*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
+	}))
 }
diff --git a/logictest/testdata/exec/filter/filter b/logictest/testdata/exec/filter/filter
index a9f597b..212cf3c 100644
--- a/logictest/testdata/exec/filter/filter
+++ b/logictest/testdata/exec/filter/filter
@@ -7,6 +7,33 @@ value2  value2  value3  null    stack1  2   2
 value3  value2  null    value4  stack1  3   3
 ----
 
+exec
+select labels, stacktrace, timestamp, value where timestamp = 2
+----
+value2  value2  value3  null    stack1  2       2
+
+exec
+select labels, stacktrace, timestamp, value where timestamp != 2
+----
+value1  value2  null    null    stack1  1       1
+value3  value2  null    value4  stack1  3       3
+
+exec
+select labels, stacktrace, timestamp, value where timestamp < 2
+----
+value1  value2  null    null    stack1  1       1
+
+exec
+select labels, stacktrace, timestamp, value where timestamp <= 2
+----
+value1  value2  null    null    stack1  1       1
+value2  value2  value3  null    stack1  2       2
+
+exec
+select labels, stacktrace, timestamp, value where timestamp > 2
+----
+value3  value2  null    value4  stack1  3       3
+
 exec
 select labels, stacktrace, timestamp, value where timestamp >= 2
 ----
diff --git a/pqarrow/arrowutils/merge_test.go b/pqarrow/arrowutils/merge_test.go
index 082f85a..3fd041e 100644
--- a/pqarrow/arrowutils/merge_test.go
+++ b/pqarrow/arrowutils/merge_test.go
@@ -35,7 +35,7 @@ func TestMerge(t *testing.T) {
 	record3 := array.NewRecord(schema, []arrow.Array{a}, int64(a.Len()))
 
 	res, err := arrowutils.MergeRecords(
-		memory.DefaultAllocator, []arrow.Record{record1, record2, record3}, []int{0},
+		memory.DefaultAllocator, []arrow.Record{record1, record2, record3}, []int{0}, 0,
 	)
 	require.NoError(t, err)
 	require.Equal(t, int64(1), res.NumCols())
@@ -47,4 +47,11 @@ func TestMerge(t *testing.T) {
 	for i := 1; i < col.Len(); i++ {
 		require.Equal(t, expected[i-1], col.Value(i))
 	}
+
+	// check that we can merge with a limit
+	res, err = arrowutils.MergeRecords(
+		memory.DefaultAllocator, []arrow.Record{record1, record2, record3}, []int{0}, 3,
+	)
+	require.NoError(t, err)
+	require.Equal(t, int64(3), res.NumRows())
 }
diff --git a/query/physicalplan/binaryscalarexpr_test.go b/query/physicalplan/binaryscalarexpr_test.go
new file mode 100644
index 0000000..96f22d2
--- /dev/null
+++ b/query/physicalplan/binaryscalarexpr_test.go
@@ -0,0 +1,53 @@
+package physicalplan
+
+import (
+	"testing"
+
+	"github.com/apache/arrow/go/v14/arrow/array"
+	"github.com/apache/arrow/go/v14/arrow/memory"
+	"github.com/apache/arrow/go/v14/arrow/scalar"
+	"github.com/stretchr/testify/require"
+
+	"github.com/polarsignals/frostdb/query/logicalplan"
+)
+
+func BenchmarkBinaryScalarOperation(b *testing.B) {
+	ab := array.NewInt64Builder(memory.DefaultAllocator)
+	for i := int64(0); i < 1_000_000; i++ {
+		ab.Append(i % 10)
+	}
+
+	arr := ab.NewInt64Array()
+	ab.Release()
+
+	s := scalar.NewInt64Scalar(4) // chosen by fair dice roll. guaranteed to be random.
+
+	operators := []logicalplan.Op{
+		logicalplan.OpAnd,
+		logicalplan.OpEq,
+		logicalplan.OpNotEq,
+		logicalplan.OpLt,
+		logicalplan.OpLtEq,
+		logicalplan.OpGt,
+		logicalplan.OpGtEq,
+	}
+
+	for _, op := range operators {
+		b.Run(op.String(), func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				_, _ = BinaryScalarOperation(arr, s, op)
+			}
+		})
+	}
+}
+
+func TestBinaryScalarOperationNotImplemented(t *testing.T) {
+	ab := array.NewInt64Builder(memory.DefaultAllocator)
+	arr := ab.NewInt64Array()
+	ab.Release()
+
+	s := scalar.NewInt64Scalar(4) // chosen by fair dice roll. guaranteed to be random.
+
+	_, err := BinaryScalarOperation(arr, s, logicalplan.OpAnd)
+	require.Equal(t, err, ErrUnsupportedBinaryOperation)
+}
diff --git a/table_test.go b/table_test.go
index 9984d39..79af2b7 100644
--- a/table_test.go
+++ b/table_test.go
@@ -74,95 +74,82 @@ func basicTable(t *testing.T, options ...Option) (*ColumnStore, *Table) {
 
 // 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 {
-		granuleSize int64
-	}{
-		"25MB": {25 * 1024 * 1024},
-		"15MB": {15 * 1024 * 1024},
-		"8MB":  {8 * 1024 * 1024},
-		"1MB":  {1024 * 1024},
+	c, table := basicTable(t)
+	defer c.Close()
+
+	generateRows := func(n int) arrow.Record {
+		rows := make(dynparquet.Samples, 0, n)
+		for i := 0; i < n; i++ {
+			rows = append(rows, dynparquet.Sample{
+				Labels: map[string]string{ // TODO would be nice to not have all the same column
+					"label1": "value1",
+					"label2": "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: rand.Int63(),
+				Value:     rand.Int63(),
+			})
+		}
+		r, err := rows.ToRecord()
+		require.NoError(t, err)
+
+		return r
 	}
 
-	for name, test := range tests {
-		t.Run(name, func(t *testing.T) {
-			c, table := basicTable(t, WithGranuleSizeBytes(test.granuleSize))
-			defer c.Close()
-
-			generateRows := func(n int) arrow.Record {
-				rows := make(dynparquet.Samples, 0, n)
-				for i := 0; i < n; i++ {
-					rows = append(rows, dynparquet.Sample{
-						Labels: map[string]string{ // TODO would be nice to not have all the same column
-							"label1": "value1",
-							"label2": "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: rand.Int63(),
-						Value:     rand.Int63(),
-					})
+	// Spawn n workers that will insert values into the table
+	maxTxID := &atomic.Uint64{}
+	n := 8
+	inserts := 100
+	rows := 10
+	wg := &sync.WaitGroup{}
+	ctx := context.Background()
+	for i := 0; i < n; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < inserts; i++ {
+				tx, err := table.InsertRecord(ctx, generateRows(rows))
+				if err != nil {
+					fmt.Println("Received error on insert: ", err)
 				}
-				r, err := rows.ToRecord()
-				require.NoError(t, err)
 
-				return r
-			}
-
-			// Spawn n workers that will insert values into the table
-			maxTxID := &atomic.Uint64{}
-			n := 8
-			inserts := 100
-			rows := 10
-			wg := &sync.WaitGroup{}
-			ctx := context.Background()
-			for i := 0; i < n; i++ {
-				wg.Add(1)
-				go func() {
-					defer wg.Done()
-					for i := 0; i < inserts; i++ {
-						tx, err := table.InsertRecord(ctx, generateRows(rows))
-						if err != nil {
-							fmt.Println("Received error on insert: ", err)
-						}
-
-						//	 Set the max tx id that we've seen
-						if maxTX := maxTxID.Load(); tx > maxTX {
-							maxTxID.CompareAndSwap(maxTX, tx)
-						}
-					}
-				}()
+				//	 Set the max tx id that we've seen
+				if maxTX := maxTxID.Load(); tx > maxTX {
+					maxTxID.CompareAndSwap(maxTX, tx)
+				}
 			}
+		}()
+	}
 
-			// Wait for all our writes to exit
-			wg.Wait()
-
-			// Wait for our last tx to be marked as complete
-			table.db.Wait(maxTxID.Load())
+	// Wait for all our writes to exit
+	wg.Wait()
 
-			pool := memory.NewGoAllocator()
+	// Wait for our last tx to be marked as complete
+	table.db.Wait(maxTxID.Load())
 
-			err := table.View(ctx, func(ctx context.Context, tx uint64) error {
-				totalrows := int64(0)
-				err := table.Iterator(
-					ctx,
-					tx,
-					pool,
-					[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
-						totalrows += ar.NumRows()
+	pool := memory.NewGoAllocator()
 
-						return nil
-					}},
-				)
+	err := table.View(ctx, func(ctx context.Context, tx uint64) error {
+		totalrows := int64(0)
+		err := table.Iterator(
+			ctx,
+			tx,
+			pool,
+			[]logicalplan.Callback{func(ctx context.Context, ar arrow.Record) error {
+				totalrows += ar.NumRows()
 
-				require.NoError(t, err)
-				require.Equal(t, int64(n*inserts*rows), totalrows)
 				return nil
-			})
-			require.NoError(t, err)
-		})
-	}
+			}},
+		)
+
+		require.NoError(t, err)
+		require.Equal(t, int64(n*inserts*rows), totalrows)
+		return nil
+	})
+	require.NoError(t, err)
 }
 
 func Benchmark_Table_Insert_1000Rows_10Iters_10Writers(b *testing.B) {
@@ -974,7 +961,6 @@ func Test_Compact_Repeated(t *testing.T) {
 
 	c, err := New(
 		WithLogger(logger),
-		WithGranuleSizeBytes(10), // NOTE: set small granule size to force compaction
 	)
 	require.NoError(t, err)
 	t.Cleanup(func() {
@@ -1101,7 +1087,7 @@ func Test_Table_DynamicColumnNotDefined(t *testing.T) {
 }
 
 func TestTableUniquePrimaryIndex(t *testing.T) {
-	c, err := New(WithGranuleSizeBytes(1), WithIndexConfig([]*IndexConfig{
+	c, err := New(WithIndexConfig([]*IndexConfig{
 		{Level: int(index.L0), MaxSize: 180, Type: CompactionTypeParquet},
 		{Level: int(index.L1), MaxSize: 1 * TiB},
 	}))