| 501 | } |
| 502 | |
| 503 | func (vlog *valueLog) rewrite(f *logFile, tr trace.Trace) error { |
| 504 | vlog.filesLock.RLock() |
| 505 | maxFid := vlog.maxFid |
| 506 | vlog.filesLock.RUnlock() |
| 507 | y.AssertTruef(uint32(f.fid) < maxFid, "fid to move: %d. Current max fid: %d", f.fid, maxFid) |
| 508 | tr.LazyPrintf("Rewriting fid: %d", f.fid) |
| 509 | |
| 510 | wb := make([]*Entry, 0, 1000) |
| 511 | var size int64 |
| 512 | |
| 513 | y.AssertTrue(vlog.db != nil) |
| 514 | var count, moved int |
| 515 | fe := func(e Entry) error { |
| 516 | count++ |
| 517 | if count%100000 == 0 { |
| 518 | tr.LazyPrintf("Processing entry %d", count) |
| 519 | } |
| 520 | |
| 521 | vs, err := vlog.db.get(e.Key) |
| 522 | if err != nil { |
| 523 | return err |
| 524 | } |
| 525 | if discardEntry(e, vs, vlog.db) { |
| 526 | return nil |
| 527 | } |
| 528 | |
| 529 | // Value is still present in value log. |
| 530 | if len(vs.Value) == 0 { |
| 531 | return errors.Errorf("Empty value: %+v", vs) |
| 532 | } |
| 533 | var vp valuePointer |
| 534 | vp.Decode(vs.Value) |
| 535 | |
| 536 | // If the entry found from the LSM Tree points to a newer vlog file, don't do anything. |
| 537 | if vp.Fid > f.fid { |
| 538 | return nil |
| 539 | } |
| 540 | // If the entry found from the LSM Tree points to an offset greater than the one |
| 541 | // read from vlog, don't do anything. |
| 542 | if vp.Offset > e.offset { |
| 543 | return nil |
| 544 | } |
| 545 | // If the entry read from LSM Tree and vlog file point to the same vlog file and offset, |
| 546 | // insert them back into the DB. |
| 547 | // NOTE: It might be possible that the entry read from the LSM Tree points to |
| 548 | // an older vlog file. See the comments in the else part. |
| 549 | if vp.Fid == f.fid && vp.Offset == e.offset { |
| 550 | moved++ |
| 551 | // This new entry only contains the key, and a pointer to the value. |
| 552 | ne := new(Entry) |
| 553 | ne.meta = 0 // Remove all bits. Different keyspace doesn't need these bits. |
| 554 | ne.UserMeta = e.UserMeta |
| 555 | ne.ExpiresAt = e.ExpiresAt |
| 556 | |
| 557 | // Create a new key in a separate keyspace, prefixed by moveKey. We are not |
| 558 | // allowed to rewrite an older version of key in the LSM tree, because then this older |
| 559 | // version would be at the top of the LSM tree. To work correctly, reads expect the |
| 560 | // latest versions to be at the top, and the older versions at the bottom. |