db: move file number assignment to just before metadata creation

Previously we preallocated all table numbers upfront based on the ingest args.
The comment on this being required in order to mark the files as pending seems
outdated, so we now assign the file number just before ingesting the sst.

Author: xinhaoz

flushable_test.go

@@ -50,15 +50,11 @@ func TestIngestedSSTFlushableAPI(t *testing.T) {
 	reset()
 
 	loadFileMeta := func(paths []string, exciseSpan KeyRange, seqNum base.SeqNum) []*manifest.TableMetadata {
-		pendingOutputs := make([]base.TableNum, len(paths))
-		for i := range paths {
-			pendingOutputs[i] = d.mu.versions.getNextTableNum()
-		}
 		jobID := d.newJobID()
 
 		// We can reuse the ingestLoad function for this test even if we're
 		// not actually ingesting a file.
-		lr, err := ingestLoad(context.Background(), d.opts, d.FormatMajorVersion(), paths, nil, nil, d.cacheHandle, &d.compressionCounters, pendingOutputs)
+		lr, err := ingestLoad(context.Background(), d.opts, d.FormatMajorVersion(), paths, nil, nil, d.cacheHandle, &d.compressionCounters, d.mu.versions.getNextDiskFileNum)
 		if err != nil {
 			t.Fatal(err)
 		}

ingest.go

@@ -450,6 +450,7 @@ func ingestLoad1(
 	}
 
 	if !meta.HasPointKeys && !meta.HasRangeKeys {
+		// Elide ingesting empty sstables.
 		return nil, keyspan.Span{}, base.BlockReadStats{}, nil
 	}
 
@@ -503,12 +504,8 @@ func ingestLoad(
 	external []ExternalFile,
 	cacheHandle *cache.Handle,
 	compressionCounters *block.CompressionCounters,
-	pending []base.TableNum,
+	getNextFileNum func() base.DiskFileNum,
 ) (ingestLoadResult, error) {
-	localFileNums := pending[:len(paths)]
-	sharedFileNums := pending[len(paths) : len(paths)+len(shared)]
-	externalFileNums := pending[len(paths)+len(shared) : len(paths)+len(shared)+len(external)]
-
 	var result ingestLoadResult
 	result.local = make([]ingestLocalMeta, 0, len(paths))
 	var lastRangeKey keyspan.Span
@@ -536,7 +533,8 @@ func ingestLoad(
 		if !shouldDisableRangeKeyChecks {
 			rangeKeyValidator = validateSuffixedBoundaries(opts.Comparer, lastRangeKey)
 		}
-		m, lastRangeKey, blockReadStats, err = ingestLoad1(ctx, opts, fmv, readable, cacheHandle, compressionCounters, localFileNums[i], rangeKeyValidator)
+		tableNum := base.TableNum(getNextFileNum())
+		m, lastRangeKey, blockReadStats, err = ingestLoad1(ctx, opts, fmv, readable, cacheHandle, compressionCounters, tableNum, rangeKeyValidator)
 		if err != nil {
 			return ingestLoadResult{}, err
 		}
@@ -561,7 +559,7 @@ func ingestLoad(
 
 	result.shared = make([]ingestSharedMeta, 0, len(shared))
 	for i := range shared {
-		m, err := ingestSynthesizeShared(opts, shared[i], sharedFileNums[i])
+		m, err := ingestSynthesizeShared(opts, shared[i], base.PhysicalTableFileNum(getNextFileNum()))
 		if err != nil {
 			return ingestLoadResult{}, err
 		}
@@ -575,7 +573,7 @@ func ingestLoad(
 	}
 	result.external = make([]ingestExternalMeta, 0, len(external))
 	for i := range external {
-		m, err := ingestLoad1External(opts, external[i], externalFileNums[i])
+		m, err := ingestLoad1External(opts, external[i], base.PhysicalTableFileNum(getNextFileNum()))
 		if err != nil {
 			return ingestLoadResult{}, err
 		}
@@ -1470,22 +1468,12 @@ func (d *DB) ingest(ctx context.Context, args ingestArgs) (IngestOperationStats,
 			}
 		}
 	}
-	// Allocate table numbers for all files being ingested and mark them as
-	// pending in order to prevent them from being deleted. Note that this causes
-	// the file number ordering to be out of alignment with sequence number
-	// ordering. The sorting of L0 tables by sequence number avoids relying on
-	// that (busted) invariant.
-	pendingOutputs := make([]base.TableNum, len(paths)+len(shared)+len(external))
-	for i := 0; i < len(paths)+len(shared)+len(external); i++ {
-		pendingOutputs[i] = d.mu.versions.getNextTableNum()
-	}
-
 	jobID := d.newJobID()
 
 	// Load the metadata for all the files being ingested. This step detects
 	// and elides empty sstables.
 	loadResult, err := ingestLoad(ctx, d.opts, d.FormatMajorVersion(), paths, shared, external,
-		d.cacheHandle, &d.compressionCounters, pendingOutputs)
+		d.cacheHandle, &d.compressionCounters, d.mu.versions.getNextDiskFileNum)
 	if err != nil {
 		return IngestOperationStats{}, err
 	}

ingest_test.go

@@ -149,7 +149,8 @@ func TestIngestLoad(t *testing.T) {
 				FS:         mem,
 			}
 			opts.WithFSDefaults()
-			lr, err := ingestLoad(context.Background(), opts, dbVersion, []string{"ext"}, nil, nil, nil, nil, []base.TableNum{1})
+			getNextFileNum := func() base.DiskFileNum { return 1 }
+			lr, err := ingestLoad(context.Background(), opts, dbVersion, []string{"ext"}, nil, nil, nil, nil, getNextFileNum)
 			if err != nil {
 				return err.Error()
 			}
@@ -184,14 +185,19 @@ func TestIngestLoadRand(t *testing.T) {
 	}
 
 	paths := make([]string, 1+rng.IntN(10))
-	pending := make([]base.TableNum, len(paths))
 	expected := make([]ingestLocalMeta, len(paths))
+	pending := make([]base.DiskFileNum, len(expected))
+	for i := range expected {
+		pending[i] = base.DiskFileNum(rng.Uint64())
+	}
+	fileNumAllocator := testFileNumAllocator{
+		fileNums: pending,
+	}
 	for i := range paths {
 		paths[i] = fmt.Sprint(i)
-		pending[i] = base.TableNum(rng.Uint64())
 		expected[i] = ingestLocalMeta{
 			TableMetadata: &manifest.TableMetadata{
-				TableNum: pending[i],
+				TableNum: base.TableNum(pending[i]),
 			},
 			path: paths[i],
 		}
@@ -248,7 +254,7 @@ func TestIngestLoadRand(t *testing.T) {
 	}
 	opts.WithFSDefaults()
 	opts.EnsureDefaults()
-	lr, err := ingestLoad(context.Background(), opts, version, paths, nil, nil, nil, nil, pending)
+	lr, err := ingestLoad(context.Background(), opts, version, paths, nil, nil, nil, nil, fileNumAllocator.nextFileNum)
 	require.NoError(t, err)
 
 	// Reset flaky stats.
@@ -273,7 +279,8 @@ func TestIngestLoadInvalid(t *testing.T) {
 		FS:       mem,
 	}
 	opts.WithFSDefaults()
-	if _, err := ingestLoad(context.Background(), opts, internalFormatNewest, []string{"invalid"}, nil, nil, nil, nil, []base.TableNum{1}); err == nil {
+	getNextFileNum := func() base.DiskFileNum { return 1 }
+	if _, err := ingestLoad(context.Background(), opts, internalFormatNewest, []string{"invalid"}, nil, nil, nil, nil, getNextFileNum); err == nil {
 		t.Fatalf("expected error, but found success")
 	}
 }
@@ -3172,3 +3179,14 @@ func runBenchmarkManySSTablesInUseKeyRanges(b *testing.B, d *DB, count int) {
 		_ = v.CalculateInuseKeyRanges(d.mu.versions.latest.l0Organizer, 0, numLevels-1, smallest, largest)
 	}
 }
+
+type testFileNumAllocator struct {
+	fileNums []base.DiskFileNum
+	idx      int
+}
+
+func (a *testFileNumAllocator) nextFileNum() base.DiskFileNum {
+	n := a.fileNums[a.idx]
+	a.idx++
+	return n
+}