writer.go

// Copyright 2011 The LevelDB-Go and Pebble Authors. All rights reserved. Use
// of this source code is governed by a BSD-style license that can be found in
// the LICENSE file.

package sstable

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"math"
	"sync"

	"github.com/cespare/xxhash/v2"
	"github.com/cockroachdb/errors"
	"github.com/cockroachdb/pebble/internal/base"
	"github.com/cockroachdb/pebble/internal/cache"
	"github.com/cockroachdb/pebble/internal/crc"
	"github.com/cockroachdb/pebble/internal/keyspan"
	"github.com/cockroachdb/pebble/internal/private"
)

// WriterMetadata holds info about a finished sstable.
type WriterMetadata struct {
	EstimatedSize  uint64
	Size           uint64
	SmallestPoint  InternalKey
	SmallestRange  InternalKey
	LargestPoint   InternalKey
	LargestRange   InternalKey
	SmallestSeqNum uint64
	LargestSeqNum  uint64
	Properties     Properties
}

func (m *WriterMetadata) updateSeqNum(seqNum uint64) {
	if m.SmallestSeqNum > seqNum {
		m.SmallestSeqNum = seqNum
	}
	if m.LargestSeqNum < seqNum {
		m.LargestSeqNum = seqNum
	}
}

// Smallest returns the smaller of SmallestPoint and SmallestRange.
func (m *WriterMetadata) Smallest(cmp Compare) InternalKey {
	if m.SmallestPoint.UserKey == nil {
		return m.SmallestRange
	}
	if m.SmallestRange.UserKey == nil {
		return m.SmallestPoint
	}
	if base.InternalCompare(cmp, m.SmallestPoint, m.SmallestRange) < 0 {
		return m.SmallestPoint
	}
	return m.SmallestRange
}

// Largest returns the larget of LargestPoint and LargestRange.
func (m *WriterMetadata) Largest(cmp Compare) InternalKey {
	if m.LargestPoint.UserKey == nil {
		return m.LargestRange
	}
	if m.LargestRange.UserKey == nil {
		return m.LargestPoint
	}
	if base.InternalCompare(cmp, m.LargestPoint, m.LargestRange) > 0 {
		return m.LargestPoint
	}
	return m.LargestRange
}

type flusher interface {
	Flush() error
}

type writeCloseSyncer interface {
	io.WriteCloser
	Sync() error
}

type checksummer struct {
	checksumType ChecksumType
	xxHasher     *xxhash.Digest
}

type averageValue struct {
	// n represents the number of values
	n uint64

	// avg represents the average value for
	// n samples.
	avg float64
}

func (a *averageValue) add(f float64) {
	oldSum := float64(a.n) * a.avg
	newSum := oldSum + f
	newAvg := newSum / float64(a.n+1)
	a.n++
	a.avg = newAvg
}

func (a *averageValue) average() float64 {
	return a.avg
}

// Writer is a table writer.
type Writer struct {
	writer    io.Writer
	bufWriter *bufio.Writer
	syncer    writeCloseSyncer
	meta      WriterMetadata
	err       error
	// cacheID and fileNum are used to remove blocks written to the sstable from
	// the cache, providing a defense in depth against bugs which cause cache
	// collisions.
	cacheID uint64
	fileNum base.FileNum
	// The following fields are copied from Options.
	blockSize               int
	blockSizeThreshold      int
	indexBlockSize          int
	indexBlockSizeThreshold int
	compare                 Compare
	split                   Split
	formatKey               base.FormatKey
	compression             Compression
	separator               Separator
	successor               Successor
	tableFormat             TableFormat
	cache                   *cache.Cache
	// disableKeyOrderChecks disables the checks that keys are added to an
	// sstable in order. It is intended for internal use only in the construction
	// of invalid sstables for testing. See tool/make_test_sstables.go.
	disableKeyOrderChecks bool
	// With two level indexes, the index/filter of a SST file is partitioned into
	// smaller blocks with an additional top-level index on them. When reading an
	// index/filter, only the top-level index is loaded into memory. The two level
	// index/filter then uses the top-level index to load on demand into the block
	// cache the partitions that are required to perform the index/filter query.
	//
	// Two level indexes are enabled automatically when there is more than one
	// index block.
	//
	// This is useful when there are very large index blocks, which generally occurs
	// with the usage of large keys. With large index blocks, the index blocks fight
	// the data blocks for block cache space and the index blocks are likely to be
	// re-read many times from the disk. The top level index, which has a much
	// smaller memory footprint, can be used to prevent the entire index block from
	// being loaded into the block cache.
	twoLevelIndex bool
	// Internal flag to allow creation of range-del-v1 format blocks. Only used
	// for testing. Note that v2 format blocks are backwards compatible with v1
	// format blocks.
	rangeDelV1Format    bool
	block               blockWriter
	indexBlock          blockWriter
	rangeDelBlock       blockWriter
	props               Properties
	propCollectors      []TablePropertyCollector
	blockPropCollectors []BlockPropertyCollector

	// todo(bananabrick) : Add some sort of acquire/release mechanism to
	// the writer.blockPropsEncoder, so that we don't accidentally reuse
	// the slice returned when blockPropsEncoder.unsafeProps is called.
	blockPropsEncoder blockPropertiesEncoder
	// compressedBuf is the destination buffer for compression. It is
	// re-used over the lifetime of the writer, avoiding the allocation of a
	// temporary buffer for each block.
	compressedBuf []byte
	// filter accumulates the filter block. If populated, the filter ingests
	// either the output of w.split (i.e. a prefix extractor) if w.split is not
	// nil, or the full keys otherwise.
	filter filterWriter
	// tmp is a scratch buffer, large enough to hold either footerLen bytes,
	// blockTrailerLen bytes, (5 * binary.MaxVarintLen64) bytes, and most
	// likely large enough for a block handle with properties.
	tmp [blockHandleLikelyMaxLen]byte

	topLevelIndexBlock blockWriter
	indexPartitions    []indexBlockWriterAndBlockProperties

	checksummer checksummer

	// parallelWriterState != nil implies that parallel compression
	// is enabled.
	parallelWriterState *parallelWriterState

	// writerMu is used to provide mutual exclusion between
	// the compaction goroutine and the writer goroutine. The
	// writerMu only has an effect if parallel compression is
	// enabled. The indexBlock, and the writer.Meta.Size is
	// accessed by both the parallel writer goroutine,
	// and writer.EstimatedSize.
	writerMu sync.Mutex
}

// parallelWriterState contains the state which is specific to parallel
// compression.
type parallelWriterState struct {
	// writeQueue is used to process compressed blocks and
	// write them to disk.
	writeQueue *writeQueue

	// compressionQueueRef holds a reference to the
	// CompressionQueueContainer.
	compressionQueueRef *CompressionQueue

	mu struct {
		sync.Mutex

		// averageBHPLength helps determine whether
		// an index block should be flushed.
		averageBHPLength averageValue

		// averageCompressionRatio is used to update writer.meta.EstimatedSize.
		averageCompressionRatio averageValue
	}

	// errCh is only used when parallel compression is enabled. Any error
	// encountered while compressing blocks in parallel or writing blocks
	// to disk is sent to this channel. errCh will be written to
	// atmost once.
	errCh chan error
}

func newParallelWriterState(writeQueue *writeQueue, compressionQueue *CompressionQueue) *parallelWriterState {
	p := &parallelWriterState{
		writeQueue:          writeQueue,
		compressionQueueRef: compressionQueue,
		errCh:               make(chan error, 1),
	}
	return p
}

func (p *parallelWriterState) updateEncodedBHPLength(len float64) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.averageBHPLength.add(len)
}

func (p *parallelWriterState) getAverageEncodedBHPLength() float64 {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.mu.averageBHPLength.average()
}

func (p *parallelWriterState) updateAverageCompressionRatio(newRatio float64) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.mu.averageCompressionRatio.add(newRatio)
}

func (p *parallelWriterState) getAverageCompressionRatio() float64 {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.mu.averageCompressionRatio.average()
}

type indexBlockWriterAndBlockProperties struct {
	writer     blockWriter
	properties []byte
}

func (w *Writer) fetchEncounteredError() error {
	if w.err != nil {
		return w.err
	}

	if w.parallelWriterState != nil && len(w.parallelWriterState.errCh) > 0 {
		w.err = <-w.parallelWriterState.errCh
	}
	return w.err
}

// Set sets the value for the given key. The sequence number is set to
// 0. Intended for use to externally construct an sstable before ingestion into
// a DB.
//
// TODO(peter): untested
func (w *Writer) Set(key, value []byte) error {
	if err := w.fetchEncounteredError(); err != nil {
		return err
	}

	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindSet), value)
}

// Delete deletes the value for the given key. The sequence number is set to
// 0. Intended for use to externally construct an sstable before ingestion into
// a DB.
//
// TODO(peter): untested
func (w *Writer) Delete(key []byte) error {
	if err := w.fetchEncounteredError(); err != nil {
		return w.err
	}

	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindDelete), nil)
}

// DeleteRange deletes all of the keys (and values) in the range [start,end)
// (inclusive on start, exclusive on end). The sequence number is set to
// 0. Intended for use to externally construct an sstable before ingestion into
// a DB.
//
// TODO(peter): untested
func (w *Writer) DeleteRange(start, end []byte) error {
	if err := w.fetchEncounteredError(); err != nil {
		return err
	}

	return w.addTombstone(base.MakeInternalKey(start, 0, InternalKeyKindRangeDelete), end)
}

// Merge adds an action to the DB that merges the value at key with the new
// value. The details of the merge are dependent upon the configured merge
// operator. The sequence number is set to 0. Intended for use to externally
// construct an sstable before ingestion into a DB.
//
// TODO(peter): untested
func (w *Writer) Merge(key, value []byte) error {
	if err := w.fetchEncounteredError(); err != nil {
		return err
	}

	return w.addPoint(base.MakeInternalKey(key, 0, InternalKeyKindMerge), value)
}

// Add adds a key/value pair to the table being written. For a given Writer,
// the keys passed to Add must be in increasing order. The exception to this
// rule is range deletion tombstones. Range deletion tombstones need to be
// added ordered by their start key, but they can be added out of order from
// point entries. Additionally, range deletion tombstones must be fragmented
// (i.e. by keyspan.Fragmenter).
func (w *Writer) Add(key InternalKey, value []byte) error {
	if err := w.fetchEncounteredError(); err != nil {
		return err
	}

	if key.Kind() == InternalKeyKindRangeDelete {
		return w.addTombstone(key, value)
	}
	return w.addPoint(key, value)
}

func (w *Writer) addPoint(key InternalKey, value []byte) error {
	if !w.disableKeyOrderChecks && w.meta.LargestPoint.UserKey != nil {
		// TODO(peter): Manually inlined version of base.InternalCompare(). This is
		// 3.5% faster on BenchmarkWriter on go1.13. Remove if go1.14 or future
		// versions show this to not be a performance win.
		x := w.compare(w.meta.LargestPoint.UserKey, key.UserKey)
		if x > 0 || (x == 0 && w.meta.LargestPoint.Trailer < key.Trailer) {
			w.err = errors.Errorf("pebble: keys must be added in order: %s, %s",
				w.meta.LargestPoint.Pretty(w.formatKey), key.Pretty(w.formatKey))
			return w.err
		}
	}

	if err := w.maybeFlush(key, value); err != nil {
		return err
	}

	for i := range w.propCollectors {
		if err := w.propCollectors[i].Add(key, value); err != nil {
			return err
		}
	}
	for i := range w.blockPropCollectors {
		if err := w.blockPropCollectors[i].Add(key, value); err != nil {
			return err
		}
	}

	w.maybeAddToFilter(key.UserKey)
	w.block.add(key, value)

	w.meta.updateSeqNum(key.SeqNum())
	// block.curKey contains the most recently added key to the block.
	w.meta.LargestPoint.UserKey = w.block.curKey[:len(w.block.curKey)-8]
	w.meta.LargestPoint.Trailer = key.Trailer
	if w.meta.SmallestPoint.UserKey == nil {
		// NB: we clone w.meta.LargestPoint rather than "key", even though they are
		// semantically identical, because we need to ensure that SmallestPoint.UserKey
		// is not nil. This is required by WriterMetadata.Smallest in order to
		// distinguish between an unset SmallestPoint and a zero-length one.
		w.meta.SmallestPoint = w.meta.LargestPoint.Clone()
	}

	w.props.NumEntries++
	switch key.Kind() {
	case InternalKeyKindDelete:
		w.props.NumDeletions++
	case InternalKeyKindMerge:
		w.props.NumMergeOperands++
	}
	w.props.RawKeySize += uint64(key.Size())
	w.props.RawValueSize += uint64(len(value))
	return nil
}

func (w *Writer) addTombstone(key InternalKey, value []byte) error {
	if !w.disableKeyOrderChecks && !w.rangeDelV1Format && w.rangeDelBlock.nEntries > 0 {
		// Check that tombstones are being added in fragmented order. If the two
		// tombstones overlap, their start and end keys must be identical.
		prevKey := base.DecodeInternalKey(w.rangeDelBlock.curKey)
		switch c := w.compare(prevKey.UserKey, key.UserKey); {
		case c > 0:
			w.err = errors.Errorf("pebble: keys must be added in order: %s, %s",
				prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
			return w.err
		case c == 0:
			prevValue := w.rangeDelBlock.curValue
			if w.compare(prevValue, value) != 0 {
				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
					(keyspan.Span{Start: prevKey, End: prevValue}).Pretty(w.formatKey),
					(keyspan.Span{Start: key, End: value}).Pretty(w.formatKey))
				return w.err
			}
			if prevKey.SeqNum() <= key.SeqNum() {
				w.err = errors.Errorf("pebble: keys must be added in order: %s, %s",
					prevKey.Pretty(w.formatKey), key.Pretty(w.formatKey))
				return w.err
			}
		default:
			prevValue := w.rangeDelBlock.curValue
			if w.compare(prevValue, key.UserKey) > 0 {
				w.err = errors.Errorf("pebble: overlapping tombstones must be fragmented: %s vs %s",
					(keyspan.Span{Start: prevKey, End: prevValue}).Pretty(w.formatKey),
					(keyspan.Span{Start: key, End: value}).Pretty(w.formatKey))
				return w.err
			}
		}
	}

	if key.Trailer == InternalKeyRangeDeleteSentinel {
		w.err = errors.Errorf("pebble: cannot add range delete sentinel: %s", key.Pretty(w.formatKey))
		return w.err
	}

	for i := range w.propCollectors {
		if err := w.propCollectors[i].Add(key, value); err != nil {
			return err
		}
	}

	w.meta.updateSeqNum(key.SeqNum())

	switch {
	case w.rangeDelV1Format:
		// Range tombstones are not fragmented in the v1 (i.e. RocksDB) range
		// deletion block format, so we need to track the largest range tombstone
		// end key as every range tombstone is added.
		//
		// Note that writing the v1 format is only supported for tests.
		if w.props.NumRangeDeletions == 0 {
			w.meta.SmallestRange = key.Clone()
			w.meta.LargestRange = base.MakeRangeDeleteSentinelKey(value).Clone()
		} else {
			if base.InternalCompare(w.compare, w.meta.SmallestRange, key) > 0 {
				w.meta.SmallestRange = key.Clone()
			}
			end := base.MakeRangeDeleteSentinelKey(value)
			if base.InternalCompare(w.compare, w.meta.LargestRange, end) < 0 {
				w.meta.LargestRange = end.Clone()
			}
		}

	default:
		// Range tombstones are fragmented in the v2 range deletion block format,
		// so the start key of the first range tombstone added will be the smallest
		// range tombstone key. The largest range tombstone key will be determined
		// in Writer.Close() as the end key of the last range tombstone added.
		if w.props.NumRangeDeletions == 0 {
			w.meta.SmallestRange = key.Clone()
		}
	}

	w.props.NumEntries++
	w.props.NumDeletions++
	w.props.NumRangeDeletions++
	w.props.RawKeySize += uint64(key.Size())
	w.props.RawValueSize += uint64(len(value))
	w.rangeDelBlock.add(key, value)
	return nil
}

func (w *Writer) maybeAddToFilter(key []byte) {
	if w.filter != nil {
		if w.split != nil {
			prefix := key[:w.split(key)]
			w.filter.addKey(prefix)
		} else {
			w.filter.addKey(key)
		}
	}
}

func (w *Writer) queueBlockForParallelCompression(key InternalKey) error {
	compressionTask := <-w.parallelWriterState.compressionQueueRef.compressionTaskBuffer
	writeTask := <-w.parallelWriterState.compressionQueueRef.writeTaskBuffer

	// We will share this channel so that the compression go routine, and the go routine which
	// writes the block to disk can communicate.
	doneCh := make(chan *compressedData, 1)

	finishedBlock := w.block.finish()
	w.meta.EstimatedSize += uint64(
		w.parallelWriterState.getAverageCompressionRatio() *
			float64(len(finishedBlock)))

	// Populate compression coordinator.
	copySliceToDst(&compressionTask.toCompress, finishedBlock)
	compressionTask.compression = w.compression
	compressionTask.checksum = w.checksummer.checksumType
	compressionTask.doneCh = doneCh

	props, err := w.getDataBlockProps()
	if err != nil {
		w.err = err
		return w.err
	}

	averageBHPLength := w.parallelWriterState.getAverageEncodedBHPLength()
	indexSep := w.indexEntrySeparator(base.DecodeInternalKey(w.block.curKey), key)
	flushIndexBlock := w.shouldFlushIndexBlock(indexSep, int(averageBHPLength))
	var indexProps []byte

	// We need to copy props here, before we call w.getIndexBlockProps
	// because the underylying slice is used by the w.blockPropsEncoder.
	copySliceToDst(&writeTask.props, copySlice(props))
	if flushIndexBlock {
		indexProps, err = w.getIndexBlockProps()
		if err != nil {
			w.err = err
			return err
		}
	}

	// Populate the rest of the fields for the writeTask.
	copySliceToDst(&writeTask.indexProps, indexProps)
	writeTask.indexSep = copyInternalKey(indexSep)
	writeTask.flushIndexBlock = flushIndexBlock
	writeTask.doneCh = doneCh

	// Queue compression + write tasks. It is important that
	// the tasks are queued in the order here.
	w.parallelWriterState.writeQueue.queue <- writeTask
	w.parallelWriterState.compressionQueueRef.queue <- compressionTask

	w.addDataBlockToIndexBlockForCollectors()
	return nil
}

func (w *Writer) maybeFlush(key InternalKey, value []byte) error {
	if !shouldFlush(key.Size(), len(value), &w.block, w.blockSize, w.blockSizeThreshold) {
		return nil
	}

	if w.parallelWriterState != nil {
		return w.queueBlockForParallelCompression(key)
	}

	// Parallel compression is disabled, so proceed as usual.
	bh, err := w.writeBlock(w.block.finish(), w.compression)
	if err != nil {
		w.err = err
		return err
	}
	props, err := w.getDataBlockProps()
	if err != nil {
		w.err = err
		return err
	}

	bhp := BlockHandleWithProperties{BlockHandle: bh, Props: props}
	indexSep := w.indexEntrySeparator(base.DecodeInternalKey(w.block.curKey), key)
	// We need to encode the bhp first, before we get the block properties for the
	// index block, because the writer.blockPropsEncoder reuses the props slice.
	encodedBHP := encodeBlockHandleWithProperties(w.tmp[:], bhp)
	flushIndexBlock := w.shouldFlushIndexBlock(indexSep, len(encodedBHP))
	var indexProps []byte
	if flushIndexBlock {
		// If we decide that we're going to flush the index block,
		// then we compute the block properties for the index block,
		// which will be used later.
		indexProps, err = w.getIndexBlockProps()
		if err != nil {
			return err
		}
	}
	if err = w.addIndexEntry(indexSep, encodedBHP, bhp, flushIndexBlock, indexProps); err != nil {
		return err
	}

	w.addDataBlockToIndexBlockForCollectors()
	return nil
}

// getBlockProps will return a byte slice of block
// properties. This slice should not be modified without copying
// and the writer.blockPropsEncoder should not be used, if the slice
// is still in use.
func (w *Writer) getDataBlockProps() ([]byte, error) {
	if len(w.blockPropCollectors) == 0 {
		return nil, nil
	}

	var err error
	w.blockPropsEncoder.resetProps()
	for i := range w.blockPropCollectors {
		scratch := w.blockPropsEncoder.getScratchForProp()
		if scratch, err = w.blockPropCollectors[i].FinishDataBlock(scratch); err != nil {
			return nil, err
		}
		if len(scratch) > 0 {
			w.blockPropsEncoder.addProp(shortID(i), scratch)
		}
	}
	return w.blockPropsEncoder.unsafeProps(), nil

}

// getIndexBlockProps will return a byte slice of block
// properties.
func (w *Writer) getIndexBlockProps() ([]byte, error) {
	w.blockPropsEncoder.resetProps()
	for i := range w.blockPropCollectors {
		scratch := w.blockPropsEncoder.getScratchForProp()
		var err error
		if scratch, err = w.blockPropCollectors[i].FinishIndexBlock(scratch); err != nil {
			return nil, err
		}
		if len(scratch) > 0 {
			w.blockPropsEncoder.addProp(shortID(i), scratch)
		}
	}
	return w.blockPropsEncoder.props(), nil
}

func (w *Writer) addDataBlockToIndexBlockForCollectors() {
	for i := range w.blockPropCollectors {
		w.blockPropCollectors[i].AddPrevDataBlockToIndexBlock()
	}
}

func (w *Writer) indexEntrySeparator(prevKey InternalKey, key InternalKey) InternalKey {
	var sep InternalKey
	if key.UserKey == nil && key.Trailer == 0 {
		sep = prevKey.Successor(w.compare, w.successor, nil)
	} else {
		sep = prevKey.Separator(w.compare, w.separator, nil, key)
	}
	return sep
}

// shouldFlushIndexBlock determines if the index block should be flushed.
// It doesn't modify any state.
func (w *Writer) shouldFlushIndexBlock(sep InternalKey, encodedBHPLength int) (flush bool) {
	w.writerMu.Lock()
	defer w.writerMu.Unlock()
	flush = supportsTwoLevelIndex(w.tableFormat) &&
		shouldFlush(
			sep.Size(), encodedBHPLength,
			&w.indexBlock, w.indexBlockSize, w.indexBlockSizeThreshold,
		)
	return flush
}

// addIndexEntry adds an index entry for the specified key and block handle.
func (w *Writer) addIndexEntry(
	sep InternalKey, encodedBHP []byte, bhp BlockHandleWithProperties, flushIndexBlock bool, indexProps []byte,
) error {
	if bhp.Length == 0 {
		// A valid blockHandle must be non-zero.
		// In particular, it must have a non-zero length.
		return nil
	}

	// If parallel compression is enabled, then we keep a weighted average value of
	// the encodedBHP length to re-use.
	if w.parallelWriterState != nil {
		w.parallelWriterState.updateEncodedBHPLength(float64(len(encodedBHP)))
	}

	if flushIndexBlock {
		w.writerMu.Lock()
		// Enable two level indexes if there is more than one index block.
		w.twoLevelIndex = true
		w.finishIndexBlock(indexProps)
		w.writerMu.Unlock()
	}

	w.writerMu.Lock()
	w.indexBlock.add(sep, encodedBHP)
	w.writerMu.Unlock()

	return nil
}

func shouldFlush(
	keySize int, valueSize int, block *blockWriter, blockSize, sizeThreshold int,
) bool {
	if block.nEntries == 0 {
		return false
	}

	size := block.estimatedSize()
	if size >= blockSize {
		return true
	}

	// The block is currently smaller than the target size.
	if size <= sizeThreshold {
		// The block is smaller than the threshold size at which we'll consider
		// flushing it.
		return false
	}

	newSize := size + keySize + valueSize
	if block.nEntries%block.restartInterval == 0 {
		newSize += 4
	}
	newSize += 4                             // varint for shared prefix length
	newSize += uvarintLen(uint32(keySize))   // varint for unshared key bytes
	newSize += uvarintLen(uint32(valueSize)) // varint for value size
	// Flush if the block plus the new entry is larger than the target size.
	return newSize > blockSize
}

// finishIndexBlock finishes the current index block and adds it to the top
// level index block. This is only used when two level indexes are enabled.
func (w *Writer) finishIndexBlock(props []byte) {
	w.indexPartitions = append(w.indexPartitions,
		indexBlockWriterAndBlockProperties{
			writer:     w.indexBlock,
			properties: props,
		})
	w.indexBlock = blockWriter{
		restartInterval: 1,
	}
}

func (w *Writer) writeTwoLevelIndex() (BlockHandle, error) {
	// Add the final unfinished index.
	indexProps, err := w.getIndexBlockProps()
	if err != nil {
		return BlockHandle{}, err
	}
	w.finishIndexBlock(indexProps)

	for i := range w.indexPartitions {
		b := &w.indexPartitions[i]
		w.props.NumDataBlocks += uint64(b.writer.nEntries)
		sep := base.DecodeInternalKey(b.writer.curKey)
		data := b.writer.finish()
		w.props.IndexSize += uint64(len(data))
		bh, err := w.writeBlock(data, w.compression)
		if err != nil {
			return BlockHandle{}, err
		}
		bhp := BlockHandleWithProperties{
			BlockHandle: bh,
			Props:       b.properties,
		}
		encoded := encodeBlockHandleWithProperties(w.tmp[:], bhp)
		w.topLevelIndexBlock.add(sep, encoded)
	}

	// NB: RocksDB includes the block trailer length in the index size
	// property, though it doesn't include the trailer in the top level
	// index size property.
	w.props.IndexPartitions = uint64(len(w.indexPartitions))
	w.props.TopLevelIndexSize = uint64(w.topLevelIndexBlock.estimatedSize())
	w.props.IndexSize += w.props.TopLevelIndexSize + blockTrailerLen

	return w.writeBlock(w.topLevelIndexBlock.finish(), w.compression)
}

// compressAndChecksum will compress the data, and return
// the compressed data.
func compressAndChecksum(
	b []byte, compression Compression, compressedBuf *[]byte, trailerBuf *[blockHandleLikelyMaxLen]byte, checksummer *checksummer) ([]byte, error) {
	// Compress the buffer, discarding the result if the improvement isn't at
	// least 12.5%.
	blockType, compressed := compressBlock(compression, b, *compressedBuf)
	if blockType != noCompressionBlockType && cap(compressed) > cap(*compressedBuf) {
		*compressedBuf = compressed[:cap(compressed)]
	}
	if len(compressed) < len(b)-len(b)/8 {
		b = compressed
	} else {
		blockType = noCompressionBlockType
	}

	(*trailerBuf)[0] = byte(blockType)
	// Calculate the checksum.
	var checksum uint32
	switch checksummer.checksumType {
	case ChecksumTypeCRC32c:
		checksum = crc.New(b).Update((*trailerBuf)[:1]).Value()
	case ChecksumTypeXXHash64:
		if checksummer.xxHasher == nil {
			checksummer.xxHasher = xxhash.New()
		} else {
			checksummer.xxHasher.Reset()
		}
		checksummer.xxHasher.Write(b)
		checksummer.xxHasher.Write((*trailerBuf)[:1])
		checksum = uint32(checksummer.xxHasher.Sum64())
	default:
		return nil, errors.Newf("unsupported checksum type: %d", checksummer.checksumType)
	}
	binary.LittleEndian.PutUint32((*trailerBuf)[1:5], checksum)

	return b, nil
}

func (w *Writer) writeBlockPostCompression(toWrite []byte, blockTrailerBuf [blockHandleLikelyMaxLen]byte) (BlockHandle, error) {
	size := w.meta.Size

	bh := BlockHandle{Offset: size, Length: uint64(len(toWrite))}

	if w.cacheID != 0 && w.fileNum != 0 {
		// Remove the block being written from the cache. This provides defense in
		// depth against bugs which cause cache collisions.
		//
		// TODO(peter): Alternatively, we could add the uncompressed value to the
		// cache.
		w.cache.Delete(w.cacheID, w.fileNum, bh.Offset)
	}

	// Write the bytes to the file.
	n, err := w.writer.Write(toWrite)
	if err != nil {
		return BlockHandle{}, err
	}
	w.meta.Size += uint64(n)
	n, err = w.writer.Write(blockTrailerBuf[:blockTrailerLen])
	if err != nil {
		return BlockHandle{}, err
	}
	w.meta.Size += uint64(n)

	return bh, nil
}

func (w *Writer) writeBlock(b []byte, compression Compression) (BlockHandle, error) {
	b, err := compressAndChecksum(b, compression, &w.compressedBuf, &w.tmp, &w.checksummer)
	if err != nil {
		return BlockHandle{}, err
	}
	return w.writeBlockPostCompression(b, w.tmp)
}

// Close finishes writing the table and closes the underlying file that the
// table was written to.
func (w *Writer) Close() (err error) {
	defer func() {
		if w.syncer == nil {
			return
		}
		err1 := w.syncer.Close()
		if err == nil {
			err = err1
		}
		w.syncer = nil
	}()

	// finish will only return after the queued up blocks have been compressed,
	// and written to disk. Note that it is important to call writeQueue.finish
	// before we check for w.err, because the write queue will set an error if it
	// finds one and we don't want to ignore it.
	if w.parallelWriterState != nil {
		w.parallelWriterState.writeQueue.finish()
	}

	if err := w.fetchEncounteredError(); err != nil {
		return err
	}

	// Finish the last data block, or force an empty data block if there
	// aren't any data blocks at all.
	if w.block.nEntries > 0 || w.indexBlock.nEntries == 0 {
		bh, err := w.writeBlock(w.block.finish(), w.compression)
		if err != nil {
			w.err = err
			return w.err
		}
		props, err := w.getDataBlockProps()
		if err != nil {
			w.err = err
			return err
		}

		bhp := BlockHandleWithProperties{BlockHandle: bh, Props: props}
		sep := w.indexEntrySeparator(base.DecodeInternalKey(w.block.curKey), InternalKey{})
		encoded := encodeBlockHandleWithProperties(w.tmp[:], bhp)
		flushIndexBlock := w.shouldFlushIndexBlock(sep, len(encoded))
		var indexProps []byte
		if flushIndexBlock {
			indexProps, err = w.getIndexBlockProps()
			if err != nil {
				w.err = err
				return err
			}
		}
		if err = w.addIndexEntry(sep, encoded, bhp, flushIndexBlock, indexProps); err != nil {
			w.err = err
			return err
		}
		w.addDataBlockToIndexBlockForCollectors()
	}

	w.props.DataSize = w.meta.Size

	// Write the filter block.
	var metaindex rawBlockWriter
	metaindex.restartInterval = 1
	if w.filter != nil {
		b, err := w.filter.finish()
		if err != nil {
			w.err = err
			return w.err
		}
		bh, err := w.writeBlock(b, NoCompression)
		if err != nil {
			w.err = err
			return w.err
		}
		n := encodeBlockHandle(w.tmp[:], bh)
		metaindex.add(InternalKey{UserKey: []byte(w.filter.metaName())}, w.tmp[:n])
		w.props.FilterPolicyName = w.filter.policyName()
		w.props.FilterSize = bh.Length
	}

	var indexBH BlockHandle
	if w.twoLevelIndex {
		w.props.IndexType = twoLevelIndex
		// Write the two level index block.
		indexBH, err = w.writeTwoLevelIndex()
		if err != nil {
			w.err = err
			return w.err
		}
	} else {
		w.props.IndexType = binarySearchIndex
		// NB: RocksDB includes the block trailer length in the index size
		// property, though it doesn't include the trailer in the filter size
		// property.
		w.props.IndexSize = uint64(w.indexBlock.estimatedSize()) + blockTrailerLen
		w.props.NumDataBlocks = uint64(w.indexBlock.nEntries)

		// Write the single level index block.
		indexBH, err = w.writeBlock(w.indexBlock.finish(), w.compression)
		if err != nil {
			w.err = err
			return w.err
		}
	}

	// Write the range-del block. The block handle must added to the meta index block
	// after the properties block has been written. This is because the entries in the
	// metaindex block must be sorted by key.
	var rangeDelBH BlockHandle
	if w.props.NumRangeDeletions > 0 {
		if !w.rangeDelV1Format {
			// Because the range tombstones are fragmented in the v2 format, the end
			// key of the last added range tombstone will be the largest range
			// tombstone key. Note that we need to make this into a range deletion
			// sentinel because sstable boundaries are inclusive while the end key of
			// a range deletion tombstone is exclusive. A Clone() is necessary as
			// rangeDelBlock.curValue is the same slice that will get passed
			// into w.writer, and some implementations of vfs.File mutate the
			// slice passed into Write(). Also, w.meta will often outlive the
			// blockWriter, and so cloning curValue allows the rangeDelBlock's
			// internal buffer to get gc'd.
			w.meta.LargestRange = base.MakeRangeDeleteSentinelKey(w.rangeDelBlock.curValue).Clone()
		}
		rangeDelBH, err = w.writeBlock(w.rangeDelBlock.finish(), NoCompression)
		if err != nil {
			w.err = err
			return w.err
		}
	}

	{
		userProps := make(map[string]string)
		for i := range w.propCollectors {
			if err := w.propCollectors[i].Finish(userProps); err != nil {
				return err
			}
		}
		for i := range w.blockPropCollectors {
			scratch := w.blockPropsEncoder.getScratchForProp()
			// Place the shortID in the first byte.
			scratch = append(scratch, byte(i))
			buf, err :=
				w.blockPropCollectors[i].FinishTable(scratch)
			if err != nil {
				return err
			}
			var prop string
			if len(buf) > 0 {
				prop = string(buf)
			}
			// NB: The property is populated in the map even if it is the
			// empty string, since the presence in the map is what indicates
			// that the block property collector was used when writing.
			userProps[w.blockPropCollectors[i].Name()] = prop
		}
		if len(userProps) > 0 {
			w.props.UserProperties = userProps
		}

		// Write the properties block.
		var raw rawBlockWriter
		// The restart interval is set to infinity because the properties block
		// is always read sequentially and cached in a heap located object. This
		// reduces table size without a significant impact on performance.
		raw.restartInterval = propertiesBlockRestartInterval
		w.props.CompressionOptions = rocksDBCompressionOptions
		w.props.save(&raw)
		bh, err := w.writeBlock(raw.finish(), NoCompression)
		if err != nil {
			w.err = err
			return w.err
		}
		n := encodeBlockHandle(w.tmp[:], bh)
		metaindex.add(InternalKey{UserKey: []byte(metaPropertiesName)}, w.tmp[:n])
	}

	// Add the range deletion block handle to the metaindex block.
	if w.props.NumRangeDeletions > 0 {
		n := encodeBlockHandle(w.tmp[:], rangeDelBH)
		// The v2 range-del block encoding is backwards compatible with the v1
		// encoding. We add meta-index entries for both the old name and the new
		// name so that old code can continue to find the range-del block and new
		// code knows that the range tombstones in the block are fragmented and
		// sorted.
		metaindex.add(InternalKey{UserKey: []byte(metaRangeDelName)}, w.tmp[:n])
		if !w.rangeDelV1Format {
			metaindex.add(InternalKey{UserKey: []byte(metaRangeDelV2Name)}, w.tmp[:n])
		}
	}

	// Write the metaindex block. It might be an empty block, if the filter
	// policy is nil. NoCompression is specified because a) RocksDB never
	// compresses the meta-index block and b) RocksDB has some code paths which
	// expect the meta-index block to not be compressed.
	metaindexBH, err := w.writeBlock(metaindex.blockWriter.finish(), NoCompression)
	if err != nil {
		w.err = err
		return w.err
	}

	// Write the table footer.
	footer := footer{
		format:      w.tableFormat,
		checksum:    w.checksummer.checksumType,
		metaindexBH: metaindexBH,
		indexBH:     indexBH,
	}
	var n int
	if n, err = w.writer.Write(footer.encode(w.tmp[:])); err != nil {
		w.err = err
		return w.err
	}

	w.meta.Size += uint64(n)
	w.meta.Properties = w.props

	// Flush the buffer.
	if w.bufWriter != nil {
		if err := w.bufWriter.Flush(); err != nil {
			w.err = err
			return err
		}
	}

	if err := w.syncer.Sync(); err != nil {
		w.err = err
		return err
	}

	// Make any future calls to Set or Close return an error.
	w.err = errors.New("pebble: writer is closed")
	return nil
}

// EstimatedSize returns the estimated size of the sstable being written if a
// called to Finish() was made without adding additional keys.
func (w *Writer) EstimatedSize() uint64 {
	var size uint64
	if w.parallelWriterState != nil {
		size = w.meta.EstimatedSize
	} else {
		size = w.meta.Size
	}

	w.writerMu.Lock()
	defer w.writerMu.Unlock()
	return size + uint64(w.block.estimatedSize()+w.indexBlock.estimatedSize())
}

// Metadata returns the metadata for the finished sstable. Only valid to call
// after the sstable has been finished.
func (w *Writer) Metadata() (*WriterMetadata, error) {
	if w.syncer != nil {
		return nil, errors.New("pebble: writer is not closed")
	}
	return &w.meta, nil
}

// WriterOption provide an interface to do work on Writer while it is being
// opened.
type WriterOption interface {
	// writerAPply is called on the writer during opening in order to set
	// internal parameters.
	writerApply(*Writer)
}

// internalTableOpt is a WriterOption that sets properties for sstables being
// created by the db itself (i.e. through flushes and compactions), as opposed
// to those meant for ingestion.
type internalTableOpt struct{}

func (i internalTableOpt) writerApply(w *Writer) {
	// Set the external sst version to 0. This is what RocksDB expects for
	// db-internal sstables; otherwise, it could apply a global sequence number.
	w.props.ExternalFormatVersion = 0
}

// ParallelCompressionOpt is a WriterOption that sets fields associated
// with parallel compression in the Writer, and sets up the writer
// to write blocks to disk asynchronously.
type ParallelCompressionOpt struct {
	CompressionQueue *CompressionQueue
	WriteQueueSize   uint64
}

func (p ParallelCompressionOpt) writerApply(w *Writer) {
	if p.CompressionQueue != nil {
		writeQueue := newWriteQueue(int(p.WriteQueueSize), w)
		w.parallelWriterState = newParallelWriterState(writeQueue, p.CompressionQueue)
	}
}

// NewWriter returns a new table writer for the file. Closing the writer will
// close the file.
func NewWriter(f writeCloseSyncer, o WriterOptions, extraOpts ...WriterOption) *Writer {
	o = o.ensureDefaults()
	w := &Writer{
		syncer: f,
		meta: WriterMetadata{
			SmallestSeqNum: math.MaxUint64,
		},
		blockSize:               o.BlockSize,
		blockSizeThreshold:      (o.BlockSize*o.BlockSizeThreshold + 99) / 100,
		indexBlockSize:          o.IndexBlockSize,
		indexBlockSizeThreshold: (o.IndexBlockSize*o.BlockSizeThreshold + 99) / 100,
		compare:                 o.Comparer.Compare,
		split:                   o.Comparer.Split,
		formatKey:               o.Comparer.FormatKey,
		compression:             o.Compression,
		separator:               o.Comparer.Separator,
		successor:               o.Comparer.Successor,
		tableFormat:             o.TableFormat,
		cache:                   o.Cache,
		block: blockWriter{
			restartInterval: o.BlockRestartInterval,
		},
		indexBlock: blockWriter{
			restartInterval: 1,
		},
		rangeDelBlock: blockWriter{
			restartInterval: 1,
		},
		topLevelIndexBlock: blockWriter{
			restartInterval: 1,
		},
	}
	w.checksummer.checksumType = o.Checksum
	if f == nil {
		w.err = errors.New("pebble: nil file")
		return w
	}

	// Note that WriterOptions are applied in two places; the ones with a
	// preApply() method are applied here, and the rest are applied after
	// default properties are set.
	type preApply interface{ preApply() }
	for _, opt := range extraOpts {
		if _, ok := opt.(preApply); ok {
			opt.writerApply(w)
		}
	}

	w.props.PrefixExtractorName = "nullptr"
	if o.FilterPolicy != nil {
		switch o.FilterType {
		case TableFilter:
			w.filter = newTableFilterWriter(o.FilterPolicy)
			if w.split != nil {
				w.props.PrefixExtractorName = o.Comparer.Name
				w.props.PrefixFiltering = true
			} else {
				w.props.WholeKeyFiltering = true
			}
		default:
			panic(fmt.Sprintf("unknown filter type: %v", o.FilterType))
		}
	}

	w.props.ColumnFamilyID = math.MaxInt32
	w.props.ComparerName = o.Comparer.Name
	w.props.CompressionName = o.Compression.String()
	w.props.MergerName = o.MergerName
	w.props.PropertyCollectorNames = "[]"
	w.props.ExternalFormatVersion = rocksDBExternalFormatVersion

	if len(o.TablePropertyCollectors) > 0 || len(o.BlockPropertyCollectors) > 0 {
		var buf bytes.Buffer
		buf.WriteString("[")
		if len(o.TablePropertyCollectors) > 0 {
			w.propCollectors = make([]TablePropertyCollector, len(o.TablePropertyCollectors))
			for i := range o.TablePropertyCollectors {
				w.propCollectors[i] = o.TablePropertyCollectors[i]()
				if i > 0 {
					buf.WriteString(",")
				}
				buf.WriteString(w.propCollectors[i].Name())
			}
		}
		if len(o.BlockPropertyCollectors) > 0 {
			// shortID is a uint8, so we cannot exceed that number of block
			// property collectors.
			if len(o.BlockPropertyCollectors) > math.MaxUint8 {
				w.err = errors.New("pebble: too many block property collectors")
				return w
			}
			// The shortID assigned to a collector is the same as its index in
			// this slice.
			w.blockPropCollectors = make([]BlockPropertyCollector, len(o.BlockPropertyCollectors))
			for i := range o.BlockPropertyCollectors {
				w.blockPropCollectors[i] = o.BlockPropertyCollectors[i]()
				if i > 0 || len(o.TablePropertyCollectors) > 0 {
					buf.WriteString(",")
				}
				buf.WriteString(w.blockPropCollectors[i].Name())
			}
		}
		buf.WriteString("]")
		w.props.PropertyCollectorNames = buf.String()
	}

	// Apply the remaining WriterOptions that do not have a preApply() method.
	for _, opt := range extraOpts {
		if _, ok := opt.(preApply); !ok {
			opt.writerApply(w)
		}
	}

	// If f does not have a Flush method, do our own buffering.
	if _, ok := f.(flusher); ok {
		w.writer = f
	} else {
		w.bufWriter = bufio.NewWriter(f)
		w.writer = w.bufWriter
	}

	return w
}

func init() {
	private.SSTableWriterDisableKeyOrderChecks = func(i interface{}) {
		w := i.(*Writer)
		w.disableKeyOrderChecks = true
	}
	private.SSTableInternalTableOpt = internalTableOpt{}
}