Merge pull request #8903 from dolthub/aaron/nbs-store-updatemanifest-…

…preopen-table-files

[no-release-notes] go/store/nbs: NomsBlockStore.UpdateManifest{,WithAppendix} pre-open table files before the manifest updates.

go/store/nbs/store.go

@@ -288,12 +288,15 @@ func (nbs *NomsBlockStore) conjoinIfRequired(ctx context.Context) (bool, error)
 	}
 }
 
-func (nbs *NomsBlockStore) UpdateManifest(ctx context.Context, updates map[hash.Hash]uint32) (mi ManifestInfo, err error) {
-	err = nbs.checkAllManifestUpdatesExist(ctx, updates)
+func (nbs *NomsBlockStore) UpdateManifest(ctx context.Context, updates map[hash.Hash]uint32) (ManifestInfo, error) {
+	chunkSources, _, err := nbs.openChunkSourcesForAddTableFiles(ctx, updates)
 	if err != nil {
-		return manifestContents{}, err
+		return manifestContents{}, nil
 	}
-	mi, _, err = nbs.updateManifestAddFiles(ctx, updates, nil, nil, nil)
+	// If these sources get added to the store, they will get cloned.
+	// Either way, we want to close these instances when we are done.
+	defer chunkSources.close()
+	mi, _, err := nbs.updateManifestAddFiles(ctx, updates, nil, nil, chunkSources)
 	return mi, err
 }
 
@@ -403,35 +406,18 @@ func (nbs *NomsBlockStore) updateManifestAddFiles(ctx context.Context, updates m
 	return updatedContents, false, nil
 }
 
-func (nbs *NomsBlockStore) UpdateManifestWithAppendix(ctx context.Context, updates map[hash.Hash]uint32, option ManifestAppendixOption) (mi ManifestInfo, err error) {
-	err = nbs.checkAllManifestUpdatesExist(ctx, updates)
+func (nbs *NomsBlockStore) UpdateManifestWithAppendix(ctx context.Context, updates map[hash.Hash]uint32, option ManifestAppendixOption) (ManifestInfo, error) {
+	chunkSources, _, err := nbs.openChunkSourcesForAddTableFiles(ctx, updates)
 	if err != nil {
-		return
+		return manifestContents{}, nil
 	}
-	mi, _, err = nbs.updateManifestAddFiles(ctx, updates, &option, nil, nil)
+	// If these sources get added to the store, they will get cloned.
+	// Either way, we want to close these instances when we are done.
+	defer chunkSources.close()
+	mi, _, err := nbs.updateManifestAddFiles(ctx, updates, &option, nil, chunkSources)
 	return mi, err
 }
 
-func (nbs *NomsBlockStore) checkAllManifestUpdatesExist(ctx context.Context, updates map[hash.Hash]uint32) error {
-	eg, ctx := errgroup.WithContext(ctx)
-	eg.SetLimit(128)
-	for h, c := range updates {
-		name := h
-		c := c
-		eg.Go(func() error {
-			ok, err := nbs.p.Exists(ctx, name.String(), c, nbs.stats)
-			if err != nil {
-				return err
-			}
-			if !ok {
-				return fmt.Errorf("missing table file referenced in UpdateManifest call: %v", h)
-			}
-			return nil
-		})
-	}
-	return eg.Wait()
-}
-
 func fromManifestAppendixOptionNewContents(upstream manifestContents, appendixSpecs []tableSpec, option ManifestAppendixOption) (manifestContents, error) {
 	contents, upstreamAppendixSpecs := upstream.removeAppendixSpecs()
 	switch option {