// compact compacts continguous groups of files that are not currently compacting.
497227
func (p *Partition) compact() {
	if p.compactionsDisabled > 0 {
		p.logger.Error("Cannot start a compaction while disabled")
		return
	}

	fs, err := p.fileSet.Duplicate()
	if err != nil {
		p.logger.Error("Attempt to compact while partition is closing", zap.Error(err))
		return
	}
	defer fs.Release()

	// Iterate over each level we are going to compact.
	// We skip the first level (0) because it is log files and they are compacted separately.
	// We skip the last level because the files have no higher level to compact into.
	minLevel, maxLevel := 1, len(p.levels)-2
	for level := minLevel; level <= maxLevel; level++ {
		// Skip level if it is currently compacting.
		if p.levelCompacting[level] {
			continue
		}

		// Collect contiguous files from the end of the level.
		files := fs.LastContiguousIndexFilesByLevel(level)
		if len(files) < 2 {
			continue
		} else if len(files) > MaxIndexMergeCount {
			files = files[len(files)-MaxIndexMergeCount:]
		}

		// We intend to do a compaction. Acquire a resource to do so.
		ref, err := p.res.Acquire()
		if err != nil {
			p.logger.Error("Attempt to compact while partition is closing", zap.Error(err))
			return
		}

		// Acquire references to the files to keep them alive through compaction.
		frefs, err := IndexFiles(files).Acquire()
		if err != nil {
			p.logger.Error("Attempt to compact a file that is closed", zap.Error(err))
			continue
		}

		// Mark the level as compacting.
		p.levelCompacting[level] = true

		// Start compacting in a separate goroutine.
		p.compactionsWG.Add(1)
		go func(level int) {
			// Compact to a new level.
			p.compactToLevel(files, frefs, level+1, ref.Closing())

			// Ensure compaction lock for the level is released.
			p.mu.Lock()
			p.levelCompacting[level] = false
			p.mu.Unlock()

			// Ensure references are released.
			frefs.Release()
			ref.Release()
			p.compactionsWG.Done()

			// Check for new compactions
			p.Compact()
		}(level)
	}
}
763829
func (lz *Lz4) Compress(in io.Reader, out io.Writer) error {
	w := lz4.NewWriter(out)
	w.Header.CompressionLevel = lz.CompressionLevel
	defer w.Close()
	_, err := io.Copy(w, in)
	return err
}
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
}
700392
func SearchCompact(ctx context.Context, requester Requester, r SearchRequest) (*CompactSearchResult, error) {
	body, err := SearchStream(SearchResponse(ctx, requester, r))
	if err != nil {
		return nil, err
	}
	return NewCompactSearchResult(body)
}
953135
func (c *Compactor) CompactFull(tsmFiles []string) ([]string, error) {
	c.mu.RLock()
	enabled := c.compactionsEnabled
	c.mu.RUnlock()

	if !enabled {
		return nil, errCompactionsDisabled
	}

	if !c.add(tsmFiles) {
		return nil, errCompactionInProgress{}
	}
	defer c.remove(tsmFiles)

	files, err := c.compact(false, tsmFiles)

	// See if we were disabled while writing a snapshot
	c.mu.RLock()
	enabled = c.compactionsEnabled
	c.mu.RUnlock()

	if !enabled {
		if err := c.removeTmpFiles(files); err != nil {
			return nil, err
		}
		return nil, errCompactionsDisabled
	}

	return files, err
}
1196001
func (d *DB) compact() {
	d.mu.Lock()
	defer d.mu.Unlock()
	if err := d.compact1(); err != nil {
		// TODO: count consecutive compaction errors and backoff.
	}
	d.compacting = false
	// The previous compaction may have produced too many files in a
	// level, so reschedule another compaction if needed.
	d.maybeScheduleCompaction()
	d.compactionCond.Broadcast()
}
935656
func (t *seriesPartitionTracker) incCompactions(status string, duration time.Duration) {
	if !t.enabled {
		return
	}

	if duration > 0 {
		labels := t.Labels()
		labels["component"] = "index"
		t.metrics.CompactionDuration.With(labels).Observe(duration.Seconds())
	}

	labels := t.Labels()
	labels["status"] = status
	t.metrics.Compactions.With(labels).Inc()
}
502768
func (x *Xz) Compress(in io.Reader, out io.Writer) error {
	w, err := xz.NewWriter(out)
	if err != nil {
		return err
	}
	defer w.Close()
	_, err = io.Copy(w, in)
	return err
}
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
}
117139
func (h *Head) compactable() bool {
	return h.MaxTime()-h.MinTime() > h.chunkRange/2*3
}
1050058
func (d *DB) maybeScheduleCompaction() {
	if d.compacting || d.closed {
		return
	}
	// TODO: check for manual compactions.
	if d.imm == nil {
		v := d.versions.currentVersion()
		// TODO: check v.fileToCompact.
		if v.compactionScore < 1 {
			// There is no work to be done.
			return
		}
	}
	d.compacting = true
	go d.compact()
}
540619
func (f *IndexFile) Compacting() bool {
	f.mu.RLock()
	v := f.compacting
	f.mu.RUnlock()
	return v
}
1173041
func (fs *FileStore) compactClientFile(orgFileName string) error {
	// Open a temporary file
	tmpFile, err := getTempFile(fs.fm.rootDir, clientsFileName)
	if err != nil {
		return err
	}
	defer func() {
		if tmpFile != nil {
			tmpFile.Close()
			os.Remove(tmpFile.Name())
		}
	}()
	bw := bufio.NewWriterSize(tmpFile, defaultBufSize)
	fileSize := int64(0)
	size := 0
	_buf := [256]byte{}
	buf := _buf[:]
	// Dump the content of active clients into the temporary file.
	for _, c := range fs.clients {
		buf, size, err = writeRecord(bw, buf, addClient, &c.ClientInfo, c.ClientInfo.Size(), fs.crcTable)
		if err != nil {
			return err
		}
		fileSize += int64(size)
	}
	// Flush the buffer on disk
	if err := bw.Flush(); err != nil {
		return err
	}
	// Start by closing the temporary file.
	if err := tmpFile.Close(); err != nil {
		return err
	}
	// Rename the tmp file to original file name
	if err := os.Rename(tmpFile.Name(), orgFileName); err != nil {
		return err
	}
	// Avoid unnecessary attempt to cleanup
	tmpFile = nil

	fs.cliDeleteRecs = 0
	fs.cliFileSize = fileSize
	fs.cliCompactTS = time.Now()
	return nil
}
517763
func NewCompactData(start xml.StartElement, decoder *xml.Decoder, delim string) (CompactData, error) {
	cd := CompactData{}
	cd.Element = start.Name.Local
	cd.Attr = map[string]string{}
	for _, a := range start.Attr {
		cd.Attr[a.Name.Local] = a.Value
	}
	err := decoder.DecodeElement(&cd, &start)
	if err != nil {
		return cd, err
	}
	if delim != "" {
		cd.Delimiter = delim
	} else if cd.CompactDelimiter.Value != "" {
		// explicitly specified delimiter, eg:  <DELIMITER value="2C"/>
		delimOrdinal, err := strconv.ParseInt(cd.CompactDelimiter.Value, 16, 8)
		if err != nil {
			return cd, fmt.Errorf("unable to parse DELIMITER value=%s", cd.CompactDelimiter.Value)
		}
		cd.Delimiter = string(rune(delimOrdinal))
	} else {
		cd.Delimiter = CompactDefaultDelim
	}
	return cd, nil
}
624125
func (opts *UniversalCompactionOptions) Destroy() {
	C.rocksdb_universal_compaction_options_destroy(opts.c)
	opts.c = nil
}
744520
func Compress(fileName, srcPath string, options *CompressOptions) error {
	if options == nil {
		options = &CompressOptions{}
	}

	srcInfo, err := os.Lstat(srcPath)
	if err != nil {
		return err
	}

	writer, err := newWriter(fileName, options)
	if err != nil {
		return err
	}

	// Removes the last slash to avoid different behaviors when `srcPath` is a folder
	srcPath = path.Clean(srcPath)

	// All files added are relative to the tar file
	// If IncludeSourceDir is true one level behind is added
	relPath := path.Dir(srcPath)
	if srcInfo.IsDir() && !options.IncludeSourceDir {
		relPath = srcPath
	}

	// To improve performance filters are prepared before.
	filters := prepareFilters(options.Filters)

	err = filepath.Walk(srcPath,
		func(filePath string, info os.FileInfo, err error) error {
			if err != nil {
				return err
			}

			// Makes the file to be relative to the tar file
			// We don't support absolute path while compressing
			// but it can be done further
			relFilePath, err := filepath.Rel(relPath, filePath)
			if err != nil {
				return err
			}

			// When IncludeSourceDir is false the relative path for the
			// root folder is '.', we have to ignore this folder
			if relFilePath == "." {
				return nil
			}

			// Check if we have to add the current file based on the user filters
			if !optimizedMatches(relFilePath, filters) {
				return nil
			}

			// All good, relative path made, filters applied, now we can write
			// the user file into tar file
			return writer.Write(filePath, relFilePath)
		})

	// If any error occurs we delete the tar file
	writer.Close(err != nil)

	return err
}
146610
func CompactRules(rules []rbacv1.PolicyRule) ([]rbacv1.PolicyRule, error) {
	compacted := make([]rbacv1.PolicyRule, 0, len(rules))

	simpleRules := map[simpleResource]*rbacv1.PolicyRule{}
	for _, rule := range rules {
		if resource, isSimple := isSimpleResourceRule(&rule); isSimple {
			if existingRule, ok := simpleRules[resource]; ok {
				// Add the new verbs to the existing simple resource rule
				if existingRule.Verbs == nil {
					existingRule.Verbs = []string{}
				}
				existingRule.Verbs = append(existingRule.Verbs, rule.Verbs...)
			} else {
				// Copy the rule to accumulate matching simple resource rules into
				simpleRules[resource] = rule.DeepCopy()
			}
		} else {
			compacted = append(compacted, rule)
		}
	}

	// Once we've consolidated the simple resource rules, add them to the compacted list
	for _, simpleRule := range simpleRules {
		compacted = append(compacted, *simpleRule)
	}

	return compacted, nil
}