<commit_before>package command

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"encoding/binary"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/gofmt"
	"github.com/golang/snappy"
	"github.com/pmylund/sortutil"
)

var snappyMagic = []byte{130, 83, 78, 65, 80, 80, 89, 0} // SNAPPY

type Segment struct {
	Ui  cli.Ui
	Cmd string

	rootPath string
	filename string
	limit    int
}

func (this *Segment) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("segment", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.rootPath, "s", "", "")
	cmdFlags.IntVar(&this.limit, "n", -1, "")
	cmdFlags.StringVar(&this.filename, "f", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if this.rootPath != "" {
		this.printSummary()
		return
	}

	if validateArgs(this, this.Ui).
		require("-f").
		invalid(args) {
		return 2
	}

	this.readSegment(this.filename)

	return
}

func (this *Segment) readSegment(filename string) {
	f, err := os.Open(filename) // readonly
	swallow(err)
	defer f.Close()

	const (
		maxKeySize = 10 << 10
		maxValSize = 2 << 20
	)

	var (
		buf = make([]byte, 12)
		key = make([]byte, maxKeySize)
		val = make([]byte, maxValSize)

		msgN        int64
		firstOffset uint64 = math.MaxUint64 // sentry
		endOffset   uint64
	)
	r := bufio.NewReader(f)
	for {
		_, err := r.Read(buf)
		if err != nil {
			if err == io.EOF {
				break
			} else {
				panic(err)
			}
		}

		// offset+size 8+4
		offset := binary.BigEndian.Uint64(buf[:8])
		size := binary.BigEndian.Uint32(buf[8:12])

		// crc32+magic+attr+keySize[key] 4+1+1+4
		r.Read(buf[0:10])

		attr := buf[5]
		keySize := binary.BigEndian.Uint32(buf[6:10])
		if keySize > 0 && keySize != math.MaxUint32 {
			_, err = r.Read(key[:keySize])
			swallow(err)
		}

		// valSize[val] 4
		_, err = r.Read(buf[:4])
		swallow(err)
		valSize := binary.BigEndian.Uint32(buf[:4])
		if valSize > 0 {
			_, err = r.Read(val[:valSize])
			swallow(err)
		}

		switch sarama.CompressionCodec(attr) {<commit_msg>comment: how kafka crc32 is calculated<commit_after>package command

import (
	"bufio"
	"bytes"
	"compress/gzip"
	"encoding/binary"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/gofmt"
	"github.com/golang/snappy"
	"github.com/pmylund/sortutil"
)

var snappyMagic = []byte{130, 83, 78, 65, 80, 80, 89, 0} // SNAPPY

type Segment struct {
	Ui  cli.Ui
	Cmd string

	rootPath string
	filename string
	limit    int
}

func (this *Segment) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("segment", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.rootPath, "s", "", "")
	cmdFlags.IntVar(&this.limit, "n", -1, "")
	cmdFlags.StringVar(&this.filename, "f", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if this.rootPath != "" {
		this.printSummary()
		return
	}

	if validateArgs(this, this.Ui).
		require("-f").
		invalid(args) {
		return 2
	}

	this.readSegment(this.filename)

	return
}

func (this *Segment) readSegment(filename string) {
	f, err := os.Open(filename) // readonly
	swallow(err)
	defer f.Close()

	const (
		maxKeySize = 10 << 10
		maxValSize = 2 << 20
	)

	var (
		buf = make([]byte, 12)
		key = make([]byte, maxKeySize)
		val = make([]byte, maxValSize)

		msgN        int64
		firstOffset uint64 = math.MaxUint64 // sentry
		endOffset   uint64
	)
	r := bufio.NewReader(f)
	for {
		_, err := r.Read(buf)
		if err != nil {
			if err == io.EOF {
				break
			} else {
				panic(err)
			}
		}

		// offset, size, crc32, magic, attrs, key-len, key-content, msg-len, msg-content
		// crc32 = crc32(magic, attrs, key-len, key-content, msg-len, msg-content)

		// offset+size 8+4
		offset := binary.BigEndian.Uint64(buf[:8])
		size := binary.BigEndian.Uint32(buf[8:12])

		// crc32+magic+attrs+keySize[key] 4+1+1+4
		r.Read(buf[0:10])

		attr := buf[5]
		keySize := binary.BigEndian.Uint32(buf[6:10])
		if keySize > 0 && keySize != math.MaxUint32 {
			_, err = r.Read(key[:keySize])
			swallow(err)
		}

		// valSize[val] 4
		_, err = r.Read(buf[:4])
		swallow(err)
		valSize := binary.BigEndian.Uint32(buf[:4])
		if valSize > 0 {
			_, err = r.Read(val[:valSize])
			swallow(err)
		}

		switch sarama.CompressionCodec(attr) {
{'fraction_non_alphanumeric': 0.17290236882886872, 'fraction_numerical': 0.032765109586008416, 'mean_word_length': 3.369439071566731, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1275983', 'n_tokens_mistral': 2085, 'n_tokens_neox': 1842, 'n_words': 365}