Optimise create table in stream_writer.go

level_handler.go

@@ -140,6 +140,29 @@ func (s *levelHandler) replaceTables(toDel, toAdd []*table.Table) error {
 	return decrRefs(toDel)
 }
 
+func (s *levelHandler) addTables(toAdd []*table.Table) error {
+	s.Lock()
+	defer s.Unlock()
+
+	// Increase totalSize first.
+	for _, t := range toAdd {
+		s.totalSize += t.Size()
+		t.IncrRef()
+		s.tables = append(s.tables, t)
+	}
+
+	return nil
+}
+
+func (s *levelHandler) sortTables() {
+	s.RLock()
+	defer s.RUnlock()
+
+	sort.Slice(s.tables, func(i, j int) bool {
+		return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
+	})
+}
+
 func decrRefs(tables []*table.Table) error {
 	for _, table := range tables {
 		if err := table.DecrRef(); err != nil {

stream_writer.go

@@ -243,6 +243,11 @@ func (sw *StreamWriter) Flush() error {
 		return err
 	}
 
+	// Sort tables at the end.
+	for _, l := range sw.db.lc.levels {
+		l.sortTables()
+	}
+
 	// Now sync the directories, so all the files are registered.
 	if sw.db.opt.ValueDir != sw.db.opt.Dir {
 		if err := syncDir(sw.db.opt.ValueDir); err != nil {
@@ -449,7 +454,7 @@ func (w *sortedWriter) createTable(builder *table.Builder) error {
 	if err := w.db.manifest.addChanges([]*pb.ManifestChange{change}); err != nil {
 		return err
 	}
-	if err := lhandler.replaceTables([]*table.Table{}, []*table.Table{tbl}); err != nil {
+	if err := lhandler.addTables([]*table.Table{tbl}); err != nil {
 		return err
 	}
 	// Release the ref held by OpenTable.