// CompactWithBatchSize removes DictionaryTerm entries with a count of zero (in batchSize batches), then
// compacts the underlying goleveldb store.  Removing entries is a workaround for github issue #374.
777946
func (ldbs *Store) CompactWithBatchSize(batchSize int) error {
	// workaround for github issue #374 - remove DictionaryTerm keys with count=0
	batch := &leveldb.Batch{}
	for {
		t, err := ldbs.db.OpenTransaction()
		if err != nil {
			return err
		}
		iter := t.NewIterator(util.BytesPrefix([]byte("d")), ldbs.defaultReadOptions)

		for iter.Next() {
			if bytes.Equal(iter.Value(), []byte{0}) {
				k := append([]byte{}, iter.Key()...)
				batch.Delete(k)
			}
			if batch.Len() == batchSize {
				break
			}
		}
		iter.Release()
		if iter.Error() != nil {
			t.Discard()
			return iter.Error()
		}

		if batch.Len() > 0 {
			err := t.Write(batch, ldbs.defaultWriteOptions)
			if err != nil {
				t.Discard()
				return err
			}
			err = t.Commit()
			if err != nil {
				return err
			}
		} else {
			t.Discard()
			break
		}
		batch.Reset()
	}

	return ldbs.db.CompactRange(util.Range{Start: nil, Limit: nil})
}
130543
func compact(ctx context.Context, client *clientv3.Client, t, rev int64) (int64, int64, error) {
	resp, err := client.KV.Txn(ctx).If(
		clientv3.Compare(clientv3.Version(compactRevKey), "=", t),
	).Then(
		clientv3.OpPut(compactRevKey, strconv.FormatInt(rev, 10)), // Expect side effect: increment Version
	).Else(
		clientv3.OpGet(compactRevKey),
	).Commit()
	if err != nil {
		return t, rev, err
	}

	curRev := resp.Header.Revision

	if !resp.Succeeded {
		curTime := resp.Responses[0].GetResponseRange().Kvs[0].Version
		return curTime, curRev, nil
	}
	curTime := t + 1

	if rev == 0 {
		// We don't compact on bootstrap.
		return curTime, curRev, nil
	}
	if _, err = client.Compact(ctx, rev); err != nil {
		return curTime, curRev, err
	}
	klog.V(4).Infof("etcd: compacted rev (%d), endpoints (%v)", rev, client.Endpoints())
	return curTime, curRev, nil
}
939308
func (d *DB) compactMemTable() error {
	meta, err := d.writeLevel0Table(d.opts.GetFileSystem(), d.imm)
	if err != nil {
		return err
	}
	err = d.versions.logAndApply(d.dirname, &versionEdit{
		logNumber: d.logNumber,
		newFiles: []newFileEntry{
			{level: 0, meta: meta},
		},
	})
	delete(d.pendingOutputs, meta.fileNum)
	if err != nil {
		return err
	}
	d.imm = nil
	d.deleteObsoleteFiles()
	return nil
}
747389
func (d *DB) compactDiskTables(c *compaction) (ve *versionEdit, pendingOutputs []uint64, retErr error) {
	defer func() {
		if retErr != nil {
			for _, fileNum := range pendingOutputs {
				delete(d.pendingOutputs, fileNum)
			}
			pendingOutputs = nil
		}
	}()

	// TODO: track snapshots.
	smallestSnapshot := d.versions.lastSequence

	// Release the d.mu lock while doing I/O.
	// Note the unusual order: Unlock and then Lock.
	d.mu.Unlock()
	defer d.mu.Lock()

	iter, err := compactionIterator(&d.tableCache, d.icmp, c)
	if err != nil {
		return nil, pendingOutputs, err
	}

	// TODO: output to more than one table, if it would otherwise be too large.
	var (
		fileNum  uint64
		filename string
		tw       *table.Writer
	)
	defer func() {
		if iter != nil {
			retErr = firstError(retErr, iter.Close())
		}
		if tw != nil {
			retErr = firstError(retErr, tw.Close())
		}
		if retErr != nil {
			d.opts.GetFileSystem().Remove(filename)
		}
	}()

	currentUkey := make([]byte, 0, 4096)
	hasCurrentUkey := false
	lastSeqNumForKey := internalKeySeqNumMax
	smallest, largest := internalKey(nil), internalKey(nil)
	for iter.Next() {
		// TODO: prioritize compacting d.imm.

		// TODO: support c.shouldStopBefore.

		ikey := internalKey(iter.Key())
		if !ikey.valid() {
			// Do not hide invalid keys.
			currentUkey = currentUkey[:0]
			hasCurrentUkey = false
			lastSeqNumForKey = internalKeySeqNumMax

		} else {
			ukey := ikey.ukey()
			if !hasCurrentUkey || d.icmp.userCmp.Compare(currentUkey, ukey) != 0 {
				// This is the first occurrence of this user key.
				currentUkey = append(currentUkey[:0], ukey...)
				hasCurrentUkey = true
				lastSeqNumForKey = internalKeySeqNumMax
			}

			drop, ikeySeqNum := false, ikey.seqNum()
			if lastSeqNumForKey <= smallestSnapshot {
				drop = true // Rule (A) referenced below.

			} else if ikey.kind() == internalKeyKindDelete &&
				ikeySeqNum <= smallestSnapshot &&
				c.isBaseLevelForUkey(d.icmp.userCmp, ukey) {

				// For this user key:
				// (1) there is no data in higher levels
				// (2) data in lower levels will have larger sequence numbers
				// (3) data in layers that are being compacted here and have
				//     smaller sequence numbers will be dropped in the next
				//     few iterations of this loop (by rule (A) above).
				// Therefore this deletion marker is obsolete and can be dropped.
				drop = true
			}

			lastSeqNumForKey = ikeySeqNum
			if drop {
				continue
			}
		}

		if tw == nil {
			d.mu.Lock()
			fileNum = d.versions.nextFileNum()
			d.pendingOutputs[fileNum] = struct{}{}
			pendingOutputs = append(pendingOutputs, fileNum)
			d.mu.Unlock()

			filename = dbFilename(d.dirname, fileTypeTable, fileNum)
			file, err := d.opts.GetFileSystem().Create(filename)
			if err != nil {
				return nil, pendingOutputs, err
			}
			tw = table.NewWriter(file, &d.icmpOpts)

			smallest = make(internalKey, len(ikey))
			copy(smallest, ikey)
			largest = make(internalKey, 0, 2*len(ikey))
		}
		largest = append(largest[:0], ikey...)
		if err := tw.Set(ikey, iter.Value(), nil); err != nil {
			return nil, pendingOutputs, err
		}
	}

	ve = &versionEdit{
		deletedFiles: map[deletedFileEntry]bool{},
		newFiles: []newFileEntry{
			{
				level: c.level + 1,
				meta: fileMetadata{
					fileNum:  fileNum,
					size:     1,
					smallest: smallest,
					largest:  largest,
				},
			},
		},
	}
	for i := 0; i < 2; i++ {
		for _, f := range c.inputs[i] {
			ve.deletedFiles[deletedFileEntry{
				level:   c.level + i,
				fileNum: f.fileNum,
			}] = true
		}
	}
	return ve, pendingOutputs, nil
}
430513
func NewShardedFlushingBatch(index ShardedIndex, maxBatchSize int) FlushingBatch {
	indices := index.shards()
	b := &shardedIndexFlushingBatch{
		batches: make([]*singleIndexFlushingBatch, len(indices)),
	}
	for i, index := range indices {
		b.batches[i] = newFlushingBatch(index, maxBatchSize)
	}
	return b
}
1324509
func (p *Partition) compactToLevel(files []*IndexFile, level int, interrupt <-chan struct{}) {
	assert(len(files) >= 2, "at least two index files are required for compaction")
	assert(level > 0, "cannot compact level zero")

	var err error
	var start time.Time

	p.tracker.IncActiveCompaction(level)
	// Set the relevant metrics at the end of any compaction.
	defer func() {
		p.mu.RLock()
		defer p.mu.RUnlock()
		p.tracker.SetFiles(uint64(len(p.fileSet.IndexFiles())), "index")
		p.tracker.SetFiles(uint64(len(p.fileSet.LogFiles())), "log")
		p.tracker.SetDiskSize(uint64(p.fileSet.Size()))
		p.tracker.DecActiveCompaction(level)

		success := err == nil
		p.tracker.CompactionAttempted(level, success, time.Since(start))
	}()

	// Build a logger for this compaction.
	log, logEnd := logger.NewOperation(p.logger, "TSI level compaction", "tsi1_compact_to_level", zap.Int("tsi1_level", level))
	defer logEnd()

	// Check for cancellation.
	select {
	case <-interrupt:
		log.Error("Cannot begin compaction", zap.Error(ErrCompactionInterrupted))
		return
	default:
	}

	// Files have already been retained by caller.
	// Ensure files are released only once.
	var once sync.Once
	defer once.Do(func() { IndexFiles(files).Release() })

	// Track time to compact.
	start = time.Now()

	// Create new index file.
	path := filepath.Join(p.path, FormatIndexFileName(p.NextSequence(), level))
	var f *os.File
	if f, err = os.Create(path); err != nil {
		log.Error("Cannot create compaction files", zap.Error(err))
		return
	}
	defer f.Close()

	log.Info("Performing full compaction",
		zap.String("src", joinIntSlice(IndexFiles(files).IDs(), ",")),
		zap.String("dst", path),
	)

	// Compact all index files to new index file.
	lvl := p.levels[level]
	var n int64
	if n, err = IndexFiles(files).CompactTo(f, p.sfile, lvl.M, lvl.K, interrupt); err != nil {
		log.Error("Cannot compact index files", zap.Error(err))
		return
	}

	// Close file.
	if err = f.Close(); err != nil {
		log.Error("Error closing index file", zap.Error(err))
		return
	}

	// Reopen as an index file.
	file := NewIndexFile(p.sfile)
	file.SetPath(path)
	if err = file.Open(); err != nil {
		log.Error("Cannot open new index file", zap.Error(err))
		return
	}

	// Obtain lock to swap in index file and write manifest.
	if err = func() error {
		p.mu.Lock()
		defer p.mu.Unlock()

		// Replace previous files with new index file.
		p.fileSet = p.fileSet.MustReplace(IndexFiles(files).Files(), file)

		// Write new manifest.
		manifestSize, err := p.Manifest().Write()
		if err != nil {
			// TODO: Close index if write fails.
			return err
		}
		p.manifestSize = manifestSize

		// Write new stats file.
		if err := p.writeStatsFile(); err != nil {
			return err
		}
		return nil
	}(); err != nil {
		log.Error("Cannot write manifest or stats", zap.Error(err))
		return
	}

	elapsed := time.Since(start)
	log.Info("Full compaction complete",
		zap.String("path", path),
		logger.DurationLiteral("elapsed", elapsed),
		zap.Int64("bytes", n),
		zap.Int("kb_per_sec", int(float64(n)/elapsed.Seconds())/1024),
	)

	// Release old files.
	once.Do(func() { IndexFiles(files).Release() })

	// Close and delete all old index files.
	for _, f := range files {
		log.Info("Removing index file", zap.String("path", f.Path()))

		if err = f.Close(); err != nil {
			log.Error("Cannot close index file", zap.Error(err))
			return
		} else if err = os.Remove(f.Path()); err != nil {
			log.Error("Cannot remove index file", zap.Error(err))
			return
		}
	}
}
1190028
func CompactRows(rows []*AddressRow) []AddressRowCompact {
	compact := make([]AddressRowCompact, 0, len(rows))
	for _, r := range rows {
		hash, err := chainhash.NewHashFromStr(r.TxHash)
		if err != nil {
			fmt.Println("Bad hash", r.TxHash)
			return nil
		}
		mhash, _ := chainhash.NewHashFromStr(r.MatchingTxHash) // zero array on error
		compact = append(compact, AddressRowCompact{
			Address:        r.Address,
			TxBlockTime:    r.TxBlockTime.UNIX(),
			MatchingTxHash: *mhash,
			TxHash:         *hash,
			TxVinVoutIndex: r.TxVinVoutIndex,
			TxType:         r.TxType,
			ValidMainChain: r.ValidMainChain,
			IsFunding:      r.IsFunding,
			Value:          r.Value,
		})
	}
	return compact
}
388349
func (db *DB) incGCSizeInBatch(batch *leveldb.Batch, change int64) (err error) {
	if change == 0 {
		return nil
	}
	gcSize, err := db.gcSize.Get()
	if err != nil {
		return err
	}

	var new uint64
	if change > 0 {
		new = gcSize + uint64(change)
	} else {
		// 'change' is an int64 and is negative
		// a conversion is needed with correct sign
		c := uint64(-change)
		if c > gcSize {
			// protect uint64 undeflow
			return nil
		}
		new = gcSize - c
	}
	db.gcSize.PutInBatch(batch, new)

	// trigger garbage collection if we reached the capacity
	if new >= db.capacity {
		db.triggerGarbageCollection()
	}
	return nil
}
1219238
func (h *DBHandle) WriteBatch(batch *UpdateBatch, sync bool) error {
	if len(batch.KVs) == 0 {
		return nil
	}
	levelBatch := &leveldb.Batch{}
	for k, v := range batch.KVs {
		key := constructLevelKey(h.dbName, []byte(k))
		if v == nil {
			levelBatch.Delete(key)
		} else {
			levelBatch.Put(key, v)
		}
	}
	if err := h.db.WriteBatch(levelBatch, sync); err != nil {
		return err
	}
	return nil
}
304066
func Compact(dst *bytes.Buffer, src []byte) error {
	return compact(dst, src, false)
}
837363
func (db *DB) CompactRange(r Range) {
	if db.closed {
		panic(ErrDBClosed)
	}

	var start, limit *C.char
	if len(r.Start) != 0 {
		start = (*C.char)(unsafe.Pointer(&r.Start[0]))
	}
	if len(r.Limit) != 0 {
		limit = (*C.char)(unsafe.Pointer(&r.Limit[0]))
	}
	C.leveldb_compact_range(
		db.Ldb, start, C.size_t(len(r.Start)), limit, C.size_t(len(r.Limit)))
}
1094614
func readUntilNullWorkSizeBatch(socketRead io.ReadCloser,
	batch []byte, workSize int) (size int, err error) {
	err = nil
	size = 0
	if workSize == 0 {
		size, err = io.ReadFull(socketRead, batch)
	} else {
		lastCheckedForNull := 0
		for err == nil {
			var offset int
			offset, err = socketRead.Read(batch[size:])
			size += offset
			if err == nil && size < len(batch) {
				endCheck := size - (size % workSize)
				if zeroWorkItem(batch[lastCheckedForNull:endCheck], workSize) {
					if size%workSize != 0 {
						rem := workSize - (size % workSize)
						offset, err = io.ReadFull(socketRead, batch[size:size+rem])
						size += offset
					}
					return
				}
				lastCheckedForNull = endCheck // need to check partial work items
			} else {
				return
			}
		}
	}
	return
}
133174
func (c *HoneycombClient) SendBatch(batch Batch) error {
	if len(batch) == 0 {
		// Nothing to send
		return nil
	}

	errs := []string{}
	for i := 0; i < len(batch); i += maxBatchSize {
		offset := i + maxBatchSize
		if offset > len(batch) {
			offset = len(batch)
		}
		if err := c.sendBatch(batch[i:offset]); err != nil {
			errs = append(errs, err.Error())
		}
	}

	if len(errs) > 0 {
		return errors.New(strings.Join(errs, "\n"))
	}

	return nil
}
915451
func (s *LambdaActivity) SetBatchSize(v int64) *LambdaActivity {
	s.BatchSize = &v
	return s
}
800698
func (c *ConfigLocal) SetBGFlushDirOpBatchSize(s int) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.bgFlushDirOpBatchSize = s
}
1298278
func (p *GenParameters) WithSize(size int) *GenParameters {
	newParameters := *p
	newParameters.MaxSize = size
	return &newParameters
}
793741
func NewTSMBatchKeyIterator(size int, fast bool, interrupt chan struct{}, readers ...*TSMReader) (KeyIterator, error) {
	var iter []*BlockIterator
	for _, r := range readers {
		iter = append(iter, r.BlockIterator())
	}

	return &tsmBatchKeyIterator{
		readers:              readers,
		values:               map[string][]Value{},
		pos:                  make([]int, len(readers)),
		size:                 size,
		iterators:            iter,
		fast:                 fast,
		buf:                  make([]blocks, len(iter)),
		mergedFloatValues:    &tsdb.FloatArray{},
		mergedIntegerValues:  &tsdb.IntegerArray{},
		mergedUnsignedValues: &tsdb.UnsignedArray{},
		mergedBooleanValues:  &tsdb.BooleanArray{},
		mergedStringValues:   &tsdb.StringArray{},
		interrupt:            interrupt,
	}, nil
}