Orphaned posting-list split parts remain live after deletion, rollup, and compaction

[KeisukeYamashita]

Describe the bug

When deletion makes a posting-list split part empty, rollup drops its reference from the parent but does not supersede the existing child record in Badger.

The logical graph reads correctly, but the unreferenced child remains a live, non-empty BitCompletePosting record. Advancing the discard timestamp and explicitly compacting the database does not reclaim its contents.

I reproduced this on upstream main at 166d22d54c75749a38ddb41da592e84be77e3434, with no implementation changes. The reproduction uses synthetic data and the normal posting-list mutation and rollup paths to create the initial splits.

To Reproduce

1. Check out the tested revision:

   git clone https://github.com/dgraph-io/dgraph.git
   cd dgraph
   git checkout --detach 166d22d54c75749a38ddb41da592e84be77e3434

2. Save the Go code below as posting/split_cleanup_test.go.

3. Run:

   go test ./posting -run '^TestRollupNativeSplitCleanup$' -count=3 -v -timeout=5m

The test creates a fresh managed Badger database and:

• Adds 1,024 UID edges with 2 KiB synthetic facets at commit timestamp 10.
• Calls rollup using the default split threshold, producing parent splits [1, 257, 513, 769].
• Deletes either the upper half of the edges or the whole list at timestamp 20.
• Calls rollup at read timestamp 30 and persists every returned KV through TxnWriter.
• Reopens the database, compacts with discard timestamp 15, and verifies historical and current reads.
• Replays the same rollup output to provide overlapping SSTs, reopens again, advances discard timestamp to 100, and calls Flatten(1).
• Scans all retained versions of child keys no longer referenced by the parent.

On unmodified main, both cases fail on each of three runs:

native split creation: edges=1024, parent splits=[1 257 513 769]
native removed split=513 retained payload=532562
Should be zero, but was 532562

native split creation: edges=1024, parent splits=[1 257 513 769]
native removed split=1 retained payload=532561
Should be zero, but was 532561

These are retained value bytes, not compressed filesystem sizes. The test stops at the first failing child in each case.

The fixture uses NumVersionsToKeep=MaxInt32, disables background compactors, and sets the L0 table threshold to 1 to exercise explicit compaction. There are no external readers or Raft state.

Reproduction and regression tests: posting/split_cleanup_test.go

/*
 * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
 * SPDX-License-Identifier: Apache-2.0
 */

package posting

import (
	"bytes"
	"context"
	"fmt"
	"math"
	"reflect"
	"testing"

	"github.com/dgraph-io/badger/v4"
	bpb "github.com/dgraph-io/badger/v4/pb"
	"github.com/dgraph-io/dgo/v250/protos/api"
	"github.com/dgraph-io/dgraph/v25/codec"
	"github.com/dgraph-io/dgraph/v25/protos/pb"
	"github.com/dgraph-io/dgraph/v25/x"
	"github.com/dgraph-io/ristretto/v2/z"
	"github.com/stretchr/testify/require"
	"google.golang.org/protobuf/proto"
)

// Each fixture owns a fresh managed database. Do not run these tests in parallel:
// reading a split uses the package-level pstore, just as the other posting tests do.
type splitCleanupFixture struct {
	db   *badger.DB
	opt  badger.Options
	key  []byte
	attr string
}

// Generate the initial split through the normal mutation and rollup implementation,
// at the default split threshold: no hand-crafted parent/child records are involved.
func (f *splitCleanupFixture) seedNative(t *testing.T) ([]uint64, []uint64) {
	t.Helper()
	var uids []uint64
	var edges []*pb.DirectedEdge
	for uid := uint64(1); uid <= 1024; uid++ {
		uids = append(uids, uid)
		edges = append(edges, &pb.DirectedEdge{Op: pb.DirectedEdge_SET, ValueId: uid,
			Facets: []*api.Facet{{Key: "label", ValType: api.Facet_STRING, Value: bytes.Repeat([]byte("x"), 2048)}},
		})
	}
	f.mutate(t, 1, 10, edges...)
	l := f.read(t, 15, uids)
	kvs, err := l.Rollup(nil, 15)
	require.NoError(t, err)
	f.write(t, kvs)
	f.reopen(t)
	l = f.read(t, 15, uids)
	splits := append([]uint64(nil), l.PartSplits()...)
	require.GreaterOrEqual(t, len(splits), 2)
	t.Logf("native split creation: edges=%d, parent splits=%v", len(uids), splits)
	return uids, splits
}

func (f *splitCleanupFixture) assertBackup(t *testing.T, ts uint64, expected []uint64) {
	t.Helper()
	l := f.read(t, ts, expected)
	buf := z.NewBuffer(10<<10, "TestSplitCleanupBackup")
	defer func() { require.NoError(t, buf.Release()) }()
	var bl pb.BackupPostingList
	_, err := l.ToBackupPostingList(&bl, nil, buf)
	require.NoError(t, err)
	pl := FromBackupPostingList(&bl)
	defer codec.FreePack(pl.Pack)
	require.Empty(t, pl.Splits)
	require.Equal(t, expected, append([]uint64{}, codec.Decode(pl.Pack, 0)...))
}

func TestRollupNativeSplitCleanup(t *testing.T) {
	for _, whole := range []bool{false, true} {
		t.Run(fmt.Sprintf("star=%t", whole), func(t *testing.T) {
			f := newSplitCleanupFixture(t, "data")
			original, splits := f.seedNative(t)
			expected := []uint64{}
			var edges []*pb.DirectedEdge
			if whole {
				edges = append(edges, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, Value: []byte(x.Star)})
			} else {
				for _, uid := range original {
					if uid >= splits[len(splits)/2] {
						edges = append(edges, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, ValueId: uid})
					} else {
						expected = append(expected, uid)
					}
				}
			}
			f.mutate(t, 19, 20, edges...)
			l := f.read(t, 30, expected)
			kvs, err := l.Rollup(nil, 30)
			require.NoError(t, err)
			f.write(t, kvs)
			f.reopen(t)
			f.db.SetDiscardTs(15)
			require.NoError(t, f.db.Flatten(1))
			f.read(t, 15, original)
			f.assertBackup(t, 15, original)
			f.assertBackup(t, 30, expected)
			latest := f.read(t, 30, expected)
			live := make(map[uint64]bool)
			for _, start := range latest.PartSplits() {
				live[start] = true
			}
			f.write(t, kvs)
			f.reopen(t)
			f.db.SetDiscardTs(100)
			require.NoError(t, f.db.Flatten(1))
			f.read(t, 100, expected)
			r := f.db.NewTransactionAt(100, false)
			defer r.Discard()
			removed := 0
			for _, start := range splits {
				if live[start] {
					continue
				}
				removed++
				key, err := x.SplitKey(f.key, start)
				require.NoError(t, err)
				it := r.NewKeyIterator(key, badger.IteratorOptions{AllVersions: true})
				var payload int64
				for it.Rewind(); it.Valid(); it.Next() {
					payload += it.Item().ValueSize()
				}
				it.Close()
				t.Logf("native removed split=%d retained payload=%d", start, payload)
				require.Zero(t, payload, "native split %d retains obsolete payload", start)
			}
			require.Positive(t, removed)
		})
	}
}

func TestRollupSplitCleanupHistoryAndReinsert(t *testing.T) {
	for _, whole := range []bool{false, true} {
		t.Run(fmt.Sprintf("star=%t", whole), func(t *testing.T) {
			f := newSplitCleanupFixture(t, "data")
			original, _ := f.seedNative(t)
			oldList := f.read(t, 15, original)
			expected := []uint64{}
			var deletes []*pb.DirectedEdge
			if whole {
				deletes = append(deletes, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, Value: []byte(x.Star)})
			} else {
				for _, uid := range original {
					if uid >= 513 {
						deletes = append(deletes, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, ValueId: uid})
					} else {
						expected = append(expected, uid)
					}
				}
			}
			f.mutate(t, 19, 20, deletes...)
			l := f.read(t, 30, expected)
			cleanup, err := l.Rollup(nil, 30)
			require.NoError(t, err)

			// A later mutation may commit before the older rollup is persisted.
			var inserts []*pb.DirectedEdge
			for _, uid := range original {
				inserts = append(inserts, &pb.DirectedEdge{Op: pb.DirectedEdge_SET, ValueId: uid,
					Facets: []*api.Facet{{Key: "label", ValType: api.Facet_STRING, Value: bytes.Repeat([]byte("y"), 2048)}},
				})
			}
			f.mutate(t, 31, 40, inserts...)

			// Exercise an already-loaded old parent while the rollup writes its KVs.
			// Its children must still be resolved at that parent's minTs, not latest.
			readErrors := make(chan error, 1)
			go func() {
				for i := 0; i < 100; i++ {
					var got []uint64
					err := oldList.Iterate(15, 0, func(p *pb.Posting) error {
						got = append(got, p.Uid)
						if len(p.Facets) != 1 || !bytes.Equal(p.Facets[0].Value, bytes.Repeat([]byte("x"), 2048)) {
							return fmt.Errorf("historical facet changed for UID %d", p.Uid)
						}
						return nil
					})
					if err != nil {
						readErrors <- err
						return
					}
					if !reflect.DeepEqual(original, got) {
						readErrors <- fmt.Errorf("historical UIDs changed")
						return
					}
				}
				readErrors <- nil
			}()
			f.write(t, cleanup)
			require.NoError(t, <-readErrors)
			f.read(t, 15, original)
			f.read(t, 30, expected)
			l = f.read(t, 40, original)
			latest, err := l.Rollup(nil, 45)
			require.NoError(t, err)
			f.write(t, latest)
			f.write(t, cleanup) // Delayed replay must not erase re-created split parts.
			f.reopen(t)
			f.db.SetDiscardTs(15)
			require.NoError(t, f.db.Flatten(1))
			f.read(t, 15, original)
			f.read(t, 19, original)
			for _, ts := range []uint64{20, 21, 30, 39} {
				f.read(t, ts, expected)
			}
			for _, ts := range []uint64{40, 41, 45} {
				l := f.read(t, ts, original)
				require.NoError(t, l.Iterate(ts, 0, func(p *pb.Posting) error {
					require.Len(t, p.Facets, 1)
					require.Equal(t, bytes.Repeat([]byte("y"), 2048), p.Facets[0].Value)
					return nil
				}))
			}
			f.assertBackup(t, 15, original)
			f.assertBackup(t, 30, expected)
			f.assertBackup(t, 45, original)
			f.write(t, latest)
			f.reopen(t)
			f.db.SetDiscardTs(100)
			require.NoError(t, f.db.Flatten(1))
			f.read(t, 100, original)
			t.Log("historical reads, concurrent old-parent reads, backup conversion, restart, and reinsertion preserved")
		})
	}
}

func newSplitCleanupFixture(t *testing.T, kind string) *splitCleanupFixture {
	t.Helper()
	f := &splitCleanupFixture{
		opt: badger.DefaultOptions(t.TempDir()).WithLogger(nil).
			WithNumVersionsToKeep(math.MaxInt32).WithNumCompactors(0).
			WithNumLevelZeroTables(1).WithNumLevelZeroTablesStall(10),
		attr: x.AttrInRootNamespace("split-cleanup"),
	}
	f.key = x.DataKey(f.attr, 1)
	switch kind {
	case "reverse":
		f.key = x.ReverseKey(f.attr, 1)
	case "index":
		f.key = x.IndexKey(f.attr, "token")
	}
	var err error
	f.db, err = badger.OpenManaged(f.opt)
	require.NoError(t, err)
	original := pstore
	pstore = f.db
	t.Cleanup(func() {
		pstore = original
		require.NoError(t, f.db.Close())
	})
	return f
}

func (f *splitCleanupFixture) reopen(t *testing.T) {
	t.Helper()
	require.NoError(t, f.db.Close())
	var err error
	f.db, err = badger.OpenManaged(f.opt)
	require.NoError(t, err)
	pstore = f.db
}

func (f *splitCleanupFixture) write(t *testing.T, kvs []*bpb.KV) {
	t.Helper()
	w := NewTxnWriter(f.db)
	require.NoError(t, w.Write(&bpb.KVList{Kv: kvs}))
	require.NoError(t, w.Flush())
}

func (f *splitCleanupFixture) seed(t *testing.T) {
	t.Helper()
	kv := MarshalPostingList(&pb.PostingList{Splits: []uint64{1, 100}}, nil)
	kv.Key, kv.Version = f.key, 10
	kvs := []*bpb.KV{kv}
	for start, uids := range map[uint64][]uint64{1: {2, 3}, 100: {100, 101}} {
		key, err := x.SplitKey(f.key, start)
		require.NoError(t, err)
		kv := MarshalPostingList(&pb.PostingList{Pack: codec.Encode(uids, blockSize)}, nil)
		kv.Key, kv.Version = key, 10
		kvs = append(kvs, kv)
	}
	f.write(t, kvs)
	f.reopen(t)
}

func (f *splitCleanupFixture) read(t *testing.T, ts uint64, expected []uint64) *List {
	t.Helper()
	l, err := readPostingListFromDisk(f.key, f.db, ts)
	require.NoError(t, err)
	require.Equal(t, expected, listToArray(t, 0, l, ts), "readTs=%d", ts)
	return l
}

func (f *splitCleanupFixture) mutate(t *testing.T, start, commit uint64, edges ...*pb.DirectedEdge) {
	t.Helper()
	l, err := readPostingListFromDisk(f.key, f.db, start)
	require.NoError(t, err)
	txn := NewTxn(start)
	for _, edge := range edges {
		edge.Entity, edge.Attr = 1, f.attr
		require.NoError(t, l.addMutation(context.Background(), txn, edge))
	}
	delta := l.mutationMap.get(start)
	require.NotNil(t, delta)
	require.NoError(t, l.commitMutation(start, commit))
	value, err := proto.Marshal(delta)
	require.NoError(t, err)
	w := NewTxnWriter(f.db)
	require.NoError(t, w.SetAt(f.key, value, BitDeltaPosting, commit))
	require.NoError(t, w.Flush())
}

// This is a regression test, not a diagnostic assertion of the broken behavior:
// it must FAIL on main because the orphan still has its original non-empty value.
func TestRollupSplitCleanupParentTimestampBoundary(t *testing.T) {
	f := newSplitCleanupFixture(t, "data")
	f.seed(t)
	f.mutate(t, 19, 20, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, Value: []byte(x.Star)})
	l := f.read(t, 30, []uint64{})
	kvs, err := l.Rollup(nil, 30)
	require.NoError(t, err)
	f.write(t, kvs)
	f.reopen(t)
	// A finite rollup writes the replacement parent at maxCommitTs+1 (21).
	// At 20 the reader still loads the original split parent and the deletion delta.
	// Its child keys must remain readable until that parent can be superseded.
	f.db.SetDiscardTs(20)
	require.NoError(t, f.db.Flatten(1))
	f.read(t, 20, []uint64{})
	f.read(t, 21, []uint64{})
}

func TestRollupReclaimsRemovedSplitParts(t *testing.T) {
	for _, kind := range []string{"data", "reverse", "index"} {
		for _, whole := range []bool{false, true} {
			for _, rollupTs := range []uint64{30, math.MaxUint64} {
				t.Run(fmt.Sprintf("%s/star=%t/readTs=%d", kind, whole, rollupTs), func(t *testing.T) {
					f := newSplitCleanupFixture(t, kind)
					f.seed(t)
					original := []uint64{2, 3, 100, 101}
					f.read(t, 15, original)
					expected := []uint64{2, 3}
					removed := []uint64{100}
					if whole {
						f.mutate(t, 19, 20, &pb.DirectedEdge{Op: pb.DirectedEdge_DEL, Value: []byte(x.Star)})
						expected = []uint64{}
						removed = []uint64{1, 100}
					} else {
						f.mutate(t, 19, 20,
							&pb.DirectedEdge{Op: pb.DirectedEdge_DEL, ValueId: 100},
							&pb.DirectedEdge{Op: pb.DirectedEdge_DEL, ValueId: 101})
					}
					l := f.read(t, 30, expected)
					kvs, err := l.Rollup(nil, rollupTs)
					require.NoError(t, err)
					f.write(t, kvs)
					f.reopen(t)

					// Compaction must retain the old graph while its timestamp is safe to read.
					f.db.SetDiscardTs(15)
					require.NoError(t, f.db.Flatten(1))
					f.read(t, 15, original)
					f.read(t, 19, original)
					for _, ts := range []uint64{20, 21, 30} {
						f.read(t, ts, expected)
					}

					// Replaying the same rollup is idempotent and gives Flatten overlapping
					// tables to compact after the discard timestamp advances.
					f.write(t, kvs)
					f.reopen(t)
					f.db.SetDiscardTs(100)
					require.NoError(t, f.db.Flatten(1))
					f.read(t, 100, expected)

					r := f.db.NewTransactionAt(100, false)
					defer r.Discard()
					for _, start := range removed {
						key, err := x.SplitKey(f.key, start)
						require.NoError(t, err)
						it := r.NewKeyIterator(key, badger.IteratorOptions{AllVersions: true})
						var payload int64
						for it.Rewind(); it.Valid(); it.Next() {
							item := it.Item()
							payload += item.ValueSize()
							t.Logf("removed split=%d version=%d meta=%d bytes=%d", start,
								item.Version(), item.UserMeta(), item.ValueSize())
						}
						it.Close()
						require.Zero(t, payload, "unreferenced split %d retains old payload after compaction", start)
					}
				})
			}
		}
	}
}

Expected behavior

Once no retained parent version needs a split part, its old contents should become reclaimable.

Cleanup must preserve historical reads while they remain valid. This report is about retained non-empty child contents, not an expectation that all empty-key metadata disappears immediately.

Screenshots

Not applicable; test output is provided above.

Environment

• OS: macOS 26.5.2, arm64
• Language/toolchain: Go 1.27.0
• Dgraph: upstream main, 166d22d54c75749a38ddb41da592e84be77e3434 (verified September 4, 2026)
• Badger: v4.9.4, as required by that revision
• Reproduction: isolated posting-package tests, not a running cluster
• Also previously reproduced on Dgraph v25.3.8

Additional context

The suspected path is:

1. removeEmptySplits removes empty parts from out.parts and the parent references. Some old parts can also be absent from the encoded output altogether.
2. List.Rollup persists only the parts still in out.parts.
3. Old child keys omitted from that output receive no empty replacement or deletion.

Relevant source: List.Rollup, removeEmptySplits.

A local candidate fix emits BitEmptyPosting for old split keys absent from the new output, using the replacement parent's timestamp, including its +1 for finite read timestamps. The existing TxnWriter then writes these with WithDiscard. Old payloads become reclaimable while zero-length markers remain.

The timestamp matters: an earlier candidate using maxCommitTs instead of the replacement parent's timestamp broke the included timestamp-boundary test after compaction.

With the corrected candidate:

• The retention tests pass for data, reverse, and index keys, for partial and whole-list deletion.
• Tests also cover historical reads, an already-loaded historical parent during rollup writes, later mutations committed before rollup persistence, delayed replay after reinsertion, database reopen, and backup posting-list conversion.
• All included tests pass with the race detector over three runs; the full posting-package suite and the focused debug-tag tests also pass.

This is not yet a full-cluster backup/restore or crash-recovery validation.

Related history:

• PR refactor: Simplify how list splits are tracked. #5920 changed empty-split handling. It touches this logic, but I have not established it as the introducing commit.
• Issue [FEATURE]: Improve typical compaction of p directories #8588 describes oversized posting directories, without identifying this specific cause.

I would be happy to contribute the fix and regression tests after confirming the intended cleanup semantics.