db: use objstorage to find obsolete objects

cleaner_test.go

@@ -16,7 +16,7 @@ import (
  "github.com/stretchr/testify/require"
 )
 
-func TestArchiveCleaner(t *testing.T) {
+func TestCleaner(t *testing.T) {
  dbs := make(map[string]*DB)
  defer func() {
  for _, db := range dbs {
@@ -26,19 +26,14 @@ func TestArchiveCleaner(t *testing.T) {
 
  mem := vfs.NewMem()
  var memLog base.InMemLogger
- opts := (&Options{
- Cleaner: ArchiveCleaner{},
- FS: vfs.WithLogging(mem, memLog.Infof),
- WALDir: "wal",
- }).WithFSDefaults()
-
+ fs := vfs.WithLogging(mem, memLog.Infof)
  datadriven.RunTest(t, "testdata/cleaner", func(t *testing.T, td *datadriven.TestData) string {
+ memLog.Reset()
  switch td.Cmd {
  case "batch":
  if len(td.CmdArgs) != 1 {
  return "batch <db>"
  }
- memLog.Reset()
  d := dbs[td.CmdArgs[0].String()]
  b := d.NewBatch()
  if err := runBatchDefineCmd(td, b); err != nil {
@@ -53,7 +48,6 @@ func TestArchiveCleaner(t *testing.T) {
  if len(td.CmdArgs) != 1 {
  return "compact <db>"
  }
- memLog.Reset()
  d := dbs[td.CmdArgs[0].String()]
  if err := d.Compact(nil, []byte("\xff"), false); err != nil {
  return err.Error()
@@ -64,13 +58,24 @@ func TestArchiveCleaner(t *testing.T) {
  if len(td.CmdArgs) != 1 {
  return "flush <db>"
  }
- memLog.Reset()
  d := dbs[td.CmdArgs[0].String()]
  if err := d.Flush(); err != nil {
  return err.Error()
  }
  return memLog.String()
 
+ case "close":
+ if len(td.CmdArgs) != 1 {
+ return "close <db>"
+ }
+ dbDir := td.CmdArgs[0].String()
+ d := dbs[dbDir]
+ if err := d.Close(); err != nil {
+ return err.Error()
+ }
+ delete(dbs, dbDir)
+ return memLog.String()
+
  case "list":
  if len(td.CmdArgs) != 1 {
  return "list <dir>"
@@ -83,26 +88,44 @@ func TestArchiveCleaner(t *testing.T) {
  return fmt.Sprintf("%s\n", strings.Join(paths, "\n"))
 
  case "open":
- if len(td.CmdArgs) != 1 && len(td.CmdArgs) != 2 {
- return "open <dir> [readonly]"
+ if len(td.CmdArgs) < 1 || len(td.CmdArgs) > 3 {
+ return "open <dir> [archive] [readonly]"
  }
- opts.ReadOnly = false
- if len(td.CmdArgs) == 2 {
- if td.CmdArgs[1].String() != "readonly" {
- return "open <dir> [readonly]"
+ dir := td.CmdArgs[0].String()
+ opts := (&Options{
+ FS: fs,
+ WALDir: dir + "_wal",
+ }).WithFSDefaults()
+
+ for i := 1; i < len(td.CmdArgs); i++ {
+ switch td.CmdArgs[i].String() {
+ case "readonly":
+ opts.ReadOnly = true
+ case "archive":
+ opts.Cleaner = ArchiveCleaner{}
+ default:
+ return "open <dir> [archive] [readonly]"
  }
- opts.ReadOnly = true
  }
-
- memLog.Reset()
- dir := td.CmdArgs[0].String()
  d, err := Open(dir, opts)
  if err != nil {
  return err.Error()
  }
  dbs[dir] = d
  return memLog.String()
 
+ case "create-bogus-file":
+ if len(td.CmdArgs) != 1 {
+ return "create-bogus-file <db/file>"
+ }
+ dst, err := fs.Create(td.CmdArgs[0].String())
+ require.NoError(t, err)
+ _, err = dst.Write([]byte("bogus data"))
+ require.NoError(t, err)
+ require.NoError(t, dst.Sync())
+ require.NoError(t, dst.Close())
+ return memLog.String()
+
  default:
  return fmt.Sprintf("unknown command: %s", td.Cmd)
  }

compaction.go

@@ -3268,20 +3268,29 @@ func (d *DB) scanObsoleteFiles(list []string) {
  }
  obsoleteOptions = append(obsoleteOptions, fi)
  case fileTypeTable:
- // TODO(radu): use the objstorage provider to find obsolete tables instead.
- if _, ok := liveFileNums[fileNum]; ok {
+ // Objects are handled through the objstorage provider below.
+ default:
+ // Don't delete files we don't know about.
+ }
+ }
+
+ objects := d.objProvider.List()
+ for _, obj := range objects {
+ switch obj.FileType {
+ case fileTypeTable:
+ if _, ok := liveFileNums[obj.FileNum]; ok {
  continue
  }
  fileMeta := &fileMetadata{
- FileNum: fileNum,
+ FileNum: obj.FileNum,
  }
- if stat, err := d.opts.FS.Stat(filename); err == nil {
- fileMeta.Size = uint64(stat.Size())
+ if size, err := d.objProvider.Size(obj); err == nil {
+ fileMeta.Size = uint64(size)
  }
  obsoleteTables = append(obsoleteTables, fileMeta)
+
  default:
- // Don't delete files we don't know about.
- continue
+ // Ignore object types we don't know about.
  }
  }
 

objstorage/provider.go

@@ -7,6 +7,7 @@ package objstorage
 import (
  "io"
  "os"
+ "sort"
  "sync"
 
  "github.com/cockroachdb/errors"
@@ -306,15 +307,6 @@ func (p *Provider) LinkOrCopyFromLocal(
  panic("unimplemented")
 }
 
-// Path returns an internal, implementation-dependent path for the object. It is
-// meant to be used for informational purposes (like logging).
-func (p *Provider) Path(meta ObjectMetadata) string {
- if !meta.IsShared() {
- return p.vfsPath(meta.FileType, meta.FileNum)
- }
- panic("unimplemented")
-}
-
 // Lookup returns the metadata of an object that is already known to the Provider.
 // Does not perform any I/O.
 func (p *Provider) Lookup(fileType base.FileType, fileNum base.FileNum) (ObjectMetadata, error) {
@@ -337,6 +329,37 @@ func (p *Provider) Lookup(fileType base.FileType, fileNum base.FileNum) (ObjectM
  return meta, nil
 }
 
+// Path returns an internal, implementation-dependent path for the object. It is
+// meant to be used for informational purposes (like logging).
+func (p *Provider) Path(meta ObjectMetadata) string {
+ if !meta.IsShared() {
+ return p.vfsPath(meta.FileType, meta.FileNum)
+ }
+ panic("unimplemented")
+}
+
+// Size returns the size of the object.
+func (p *Provider) Size(meta ObjectMetadata) (int64, error) {
+ if !meta.IsShared() {
+ return p.vfsSize(meta.FileType, meta.FileNum)
+ }
+ panic("unimplemented")
+}
+
+// List returns the objects currently known to the provider. Does not perform any I/O.
+func (p *Provider) List() []ObjectMetadata {
+ p.mu.RLock()
+ defer p.mu.RUnlock()
+ res := make([]ObjectMetadata, len(p.mu.knownObjects))
+ for _, meta := range p.mu.knownObjects {
+ res = append(res, meta)
+ }
+ sort.Slice(res, func(i, j int) bool {
+ return res[i].FileNum < res[j].FileNum
+ })
+ return res
+}
+
 func (p *Provider) addMetadata(meta ObjectMetadata) {
  p.mu.Lock()
  defer p.mu.Unlock()

objstorage/vfs.go

@@ -63,3 +63,12 @@ func (p *Provider) vfsFindExisting(ls []string) []ObjectMetadata {
  }
  return res
 }
+
+func (p *Provider) vfsSize(fileType base.FileType, fileNum base.FileNum) (int64, error) {
+ filename := p.vfsPath(fileType, fileNum)
+ stat, err := p.st.FS.Stat(filename)
+ if err != nil {
+ return 0, err
+ }
+ return stat.Size(), nil
+}