feat(binlog): 增加 logical clock 元数据解析与统计支撑

- 暴露事务级 last_committed、sequence_number、transaction_length 和 commit timestamp - 在 GTID event 转换时透传 logical clock 元数据 - 新增 ParseOptions、ParseProgress，支持上下文取消和解析进度回调 - 保留 TransactionPayloadEvent 展开后的 tablemap 与压缩类型信息 - 增加 TransactionSummary 辅助结构，便于上层统计事务结果、耗时、表分布和逻辑时钟 - 清理测试对外部大 binlog 样本的依赖，保证独立库测试可运行 - 修正跨平台测试与 GTID 输出格式断言，提升发版稳定性
refactor(parse): 拆分 parse.go 并修复事务/过滤一致性问题
2026-05-10 14:02:53 +08:00 · 2026-03-19 17:04:35 +08:00 · 2026-03-12 22:30:15 +08:00
16 changed files with 2057 additions and 926 deletions
@@ -3,10 +3,19 @@ package binlog
 import (
 	"os"
 	"os/exec"
+	"runtime"
 	"testing"
 )

+func requireBundledMysqlbinlog(t *testing.T) {
+	t.Helper()
+	if runtime.GOOS != "darwin" || runtime.GOARCH != "amd64" {
+		t.Skip("skips bundled mysqlbinlog validation: test/mysqlbinlog is a Darwin amd64 executable")
+	}
+}
+
 func TestDumpBinlogFromPos0(t *testing.T) {
+	requireBundledMysqlbinlog(t)
 	defer os.Remove("./test/test-mysql-bin-dump")
 	if err := DumpBinlogFromPos("./test/test-mysql-bin", 107, "./test/test-mysql-bin-dump"); nil != err {
 		t.Errorf("expect no err, but got %v", err)
@@ -17,6 +26,7 @@ func TestDumpBinlogFromPos0(t *testing.T) {
 }

 func TestDumpBinlogFromPos1(t *testing.T) {
+	requireBundledMysqlbinlog(t)
 	defer os.Remove("./test/test-mysql-bin-dump")
 	if err := DumpBinlogFromPos("./test/test-mysql-bin", 24959, "./test/test-mysql-bin-dump"); nil != err {
 		t.Errorf("expect no err, but got %v", err)
@@ -27,6 +37,7 @@ func TestDumpBinlogFromPos1(t *testing.T) {
 }

 func TestDumpUnexecutedBinlogByGtid(t *testing.T) {
+	requireBundledMysqlbinlog(t)
 	defer os.Remove("./test/test-mysql-bin-dump")
 	if err := DumpUnexecutedBinlogByGtid("./test/mysql-bin56.000003", "f60ab33c-c604-11e3-8e1c-e66ccf50db66:1-73", "./test/test-mysql-bin-dump", false); nil != err {
 		t.Errorf("expect no err, but got %v", err)
@@ -37,6 +48,7 @@ func TestDumpUnexecutedBinlogByGtid(t *testing.T) {
 }

 func TestDumpBinlogWithOnlyHeader(t *testing.T) {
+	requireBundledMysqlbinlog(t)
 	defer os.Remove("./test/test-mysql-bin-dump")
 	if err := DumpBinlogFromPos("./test/only-header-mysql-bin", 231, "./test/test-mysql-bin-dump"); nil != err {
 		t.Errorf("expect no err, but got %v", err)
@@ -1,15 +1,13 @@
 package binlog

-import (
-	"testing"
-)
+import "testing"

 func TestGetAllGtidOfBinlogDir(t *testing.T) {
 	desc, err := GetAllGtidOfBinlogDir("./test", "mysql-bin56")
 	if nil != err {
 		t.Fatalf("unexpected error %v", err)
 	}
-	if "7E23401AC60311E38E135E10E6A05CFB:1-6,8186FC1EC5FF11E38DF9E66CCF50DB66:1-11,A6CE328CC60211E38E0DE66CCF50DB66:1-6,B7009920C60111E38E075E10E6A05CFB:1-6,F60AB33CC60411E38E1CE66CCF50DB66:1-136" != desc {
+	if "7E23401AC60311E38E135E10E6A05CFB:1-6,8186FC1EC5FF11E38DF9E66CCF50DB66:1-11,A6CE328CC60211E38E0DE66CCF50DB66:1-6,B7009920C60111E38E075E10E6A05CFB:1-6,F60AB33CC60411E38E1CE66CCF50DB66:1-136" != normalizeGtidForTest(desc) {
 		t.Fatalf("wrong gtid %v", desc)
 	}
 }
@@ -1,6 +1,7 @@
 package binlog

 import (
+	"strings"
 	"testing"
 )

@@ -9,7 +10,21 @@ func TestGetGtidOfBinlog(t *testing.T) {
 	if nil != err {
 		t.Fatalf("unexpected error %v", err)
 	}
-	if "F60AB33CC60411E38E1CE66CCF50DB66:1-136" != desc {
+	if "F60AB33CC60411E38E1CE66CCF50DB66:1-136" != normalizeGtidForTest(desc) {
 		t.Fatalf("wrong gtid %v", desc)
 	}
 }
+
+func normalizeGtidForTest(desc string) string {
+	parts := strings.Split(desc, ",")
+	for i, part := range parts {
+		gtidParts := strings.SplitN(part, ":", 2)
+		if len(gtidParts) != 2 {
+			parts[i] = strings.ToUpper(part)
+			continue
+		}
+		uuid := strings.ReplaceAll(strings.ToUpper(gtidParts[0]), "-", "")
+		parts[i] = uuid + ":" + strings.ToUpper(gtidParts[1])
+	}
+	return strings.Join(parts, ",")
+}
@@ -1,923 +1,8 @@
 package binlog

-import (
-	"b612.me/mysql/gtid"
-	"b612.me/staros"
-	"bufio"
-	"bytes"
-	"errors"
-	"fmt"
-	"github.com/starainrt/go-mysql/replication"
-	"io"
-	"os"
-	"strings"
-	"sync"
-	"time"
-)
-
-var (
-	ErrInvalidBinlogHeader = errors.New("invalid binlog file header")
-	ErrEventTooSmall       = errors.New("event size too small")
-)
-
-const (
-	CompressionNone uint64 = 255
-	CompressionZSTD uint64 = 0
-)
-
-const (
-	maxPooledBodyCap   = 4 << 20 // 4MB
-	defaultReadBufSize = 1 << 20 // 1MB
-)
-
-type TxDetail struct {
-	StartPos        int             `json:"startPos"`
-	EndPos          int             `json:"endPos"`
-	RowCount        int             `json:"rowCount"`
-	Timestamp       int64           `json:"timestamp"`
-	Time            time.Time       `json:"time"`
-	Sql             string          `json:"sql"`
-	Db              string          `json:"db"`
-	Table           string          `json:"table"`
-	SqlType         string          `json:"sqlType"`
-	CompressionType string          `json:"compressionType"`
-	Rows            [][]interface{} `json:"rows"`
-}
-
-const (
-	STATUS_PREPARE uint8 = iota
-	STATUS_BEGIN
-	STATUS_COMMIT
-	STATUS_ROLLBACK
-)
-
-type Transaction struct {
-	GTID          string     `json:"gtid"`
-	Timestamp     int64      `json:"timestamp"`
-	Time          time.Time  `json:"time"`
-	StartPos      int        `json:"startPos"`
-	EndPos        int        `json:"endPos"`
-	Size          int        `json:"size"`
-	RowsCount     int        `json:"rowsCount"`
-	Status        uint8      `json:"status"`
-	TxStartTime   int64      `json:"txStartTime"`
-	TxEndTime     int64      `json:"txEndTime"`
-	sqlOrigin     []string   `json:"sqlOrigin"`
-	Txs           []TxDetail `json:"txs"`
-	dmlEventCount int
-}
-
-func (t Transaction) GetSqlOrigin() []string {
-	return t.sqlOrigin
-}
-
-type BinlogFilter struct {
-	IncludeGtid      string
-	ExcludeGtid      string
-	IncludeTables    []string
-	ExcludeTables    []string
-	StartPos         int
-	EndPos           int
-	StartDate        time.Time
-	EndDate          time.Time
-	BigThan          int
-	SmallThan        int
-	OnlyShowGtid     bool
-	OnlyShowDML      bool
-	PickTxAllIfMatch bool
-	ExcludeBlank     bool
-	IncludeBlank     bool
-}
-
-type BinlogEvent struct {
-	Type            string
-	DB              string
-	TB              string
-	Data            string
-	RowCnt          uint32
-	Rows            [][]interface{}
-	CompressionType string
-}
-
-type tableMatcher struct {
-	exactMatch map[string]bool
-	dbWildcard map[string]bool
-	tbWildcard map[string]bool
-	matchAll   bool
-}
-
-func (m *tableMatcher) match(db, tb string) bool {
-	if m.matchAll {
-		return true
-	}
-	if m.exactMatch[db+"."+tb] {
-		return true
-	}
-	if m.dbWildcard[db] {
-		return true
-	}
-	if m.tbWildcard[tb] {
-		return true
-	}
-	return false
-}
-
-var bodyBufPool = sync.Pool{
-	New: func() any {
-		b := make([]byte, 0, 64*1024)
-		return &b
-	},
-}
-
-func getBodyBuf(n int) []byte {
-	p := bodyBufPool.Get().(*[]byte)
-	if cap(*p) < n {
-		b := make([]byte, n)
-		return b
-	}
-	return (*p)[:n]
-}
-
-func putBodyBuf(b []byte) {
-	if cap(b) > maxPooledBodyCap {
-		return
-	}
-	b = b[:0]
-	bodyBufPool.Put(&b)
-}
-
-func ParseBinlogFile(path string, fx func(transaction Transaction) bool) error {
-	return parseOneBinlog(path, fx)
-}
-
-func parseOneBinlog(path string, fx func(Transaction) bool) error {
-	if !staros.Exists(path) {
-		return os.ErrNotExist
-	}
-	f, err := os.Open(path)
-	if err != nil {
-		return err
-	}
-	defer f.Close()
-
-	if err := validateBinlogHeader(f); err != nil {
-		return err
-	}
-
-	br := bufio.NewReaderSize(f, defaultReadBufSize)
-	return parseBinlogDetail(br, fx)
-}
-
-func validateBinlogHeader(f *os.File) error {
-	const fileTypeBytes = int64(4)
-	b := make([]byte, fileTypeBytes)
-
-	if _, err := f.Read(b); err != nil {
-		return fmt.Errorf("read binlog header failed: %w", err)
-	}
-	if !bytes.Equal(b, replication.BinLogFileHeader) {
-		return ErrInvalidBinlogHeader
-	}
-	if _, err := f.Seek(fileTypeBytes, io.SeekStart); err != nil {
-		return fmt.Errorf("seek after header failed: %w", err)
-	}
-	return nil
-}
-
-func readEventHeader(r io.Reader, parser *replication.BinlogParser, headBuf []byte) (*replication.EventHeader, error) {
-	if _, err := io.ReadFull(r, headBuf); err != nil {
-		return nil, err
-	}
-	h, err := parser.ParseHeader(headBuf)
-	if err != nil {
-		return nil, fmt.Errorf("parse header failed: %w", err)
-	}
-	if h.EventSize <= uint32(replication.EventHeaderSize) {
-		return nil, fmt.Errorf("%w: event size is %d", ErrEventTooSmall, h.EventSize)
-	}
-	return h, nil
-}
-
-func readEventBody(r io.Reader, h *replication.EventHeader) ([]byte, error) {
-	bodyLen := int(h.EventSize) - replication.EventHeaderSize
-	body := getBodyBuf(bodyLen)
-	if _, err := io.ReadFull(r, body); err != nil {
-		putBodyBuf(body)
-		return nil, fmt.Errorf("read event body failed: %w (need %d bytes)", err, bodyLen)
-	}
-	return body, nil
-}
-
-func skipEventBody(r io.Reader, h *replication.EventHeader) error {
-	bodyLen := int64(h.EventSize) - int64(replication.EventHeaderSize)
-	if bodyLen <= 0 {
-		return nil
-	}
-	if _, err := io.CopyN(io.Discard, r, bodyLen); err != nil {
-		return fmt.Errorf("skip event body failed: %w", err)
-	}
-	return nil
-}
-
-func parseEvent(parser *replication.BinlogParser, h *replication.EventHeader, body []byte) (replication.Event, error) {
-	e, err := parser.ParseEvent(h, body, nil)
-	if err != nil {
-		return nil, fmt.Errorf("parse event failed at pos %d: %w", h.LogPos, err)
-	}
-	return e, nil
-}
-
-func finalizeTx(tx *Transaction, onlyShowGtid bool) {
-	idx := 0
-	for k, v := range tx.Txs {
-		if v.SqlType != "query" && len(tx.sqlOrigin) > idx {
-			v.Sql = tx.sqlOrigin[idx]
-			idx++
-		}
-		tx.RowsCount += v.RowCount
-		tx.Txs[k] = v
-	}
-
-	if onlyShowGtid {
-		tx.Size = 0
-	} else {
-		tx.Size = tx.EndPos - tx.StartPos
-	}
-}
-
-func fillTimeLazy(tx *Transaction) {
-	if tx.Timestamp != 0 && tx.Time.IsZero() {
-		tx.Time = time.Unix(tx.Timestamp, 0)
-	}
-	for i := range tx.Txs {
-		if tx.Txs[i].Timestamp != 0 && tx.Txs[i].Time.IsZero() {
-			tx.Txs[i].Time = time.Unix(tx.Txs[i].Timestamp, 0)
-		}
-	}
-}
-
-func parseBinlogDetail(r io.Reader, f func(Transaction) bool) error {
-	parser := replication.NewBinlogParser()
-	parser.SetParseTime(false)
-	parser.SetUseDecimal(false)
-
-	var (
-		tbMapPos uint32
-		tx       Transaction
-		headBuf  = make([]byte, replication.EventHeaderSize)
-	)
-	currentGtid := ""
-
-	for {
-		h, err := readEventHeader(r, parser, headBuf)
-		if err == io.EOF {
-			if currentGtid != "" {
-				finalizeTx(&tx, false)
-				fillTimeLazy(&tx)
-				if f != nil {
-					f(tx)
-				}
-			}
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-
-		body, err := readEventBody(r, h)
-		if err != nil {
-			return err
-		}
-
-		e, err := parseEvent(parser, h, body)
-		putBodyBuf(body)
-		if err != nil {
-			return err
-		}
-
-		if h.EventType == replication.TABLE_MAP_EVENT {
-			tbMapPos = h.LogPos - h.EventSize
-		}
-
-		evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
-		for _, ev := range evs {
-			startPos := 0
-			if ev.Type == "query" || ev.Type == "gtid" {
-				startPos = int(h.LogPos - h.EventSize)
-			} else {
-				startPos = int(tbMapPos)
-			}
-
-			switch ev.Type {
-			case "gtid":
-				if currentGtid != "" {
-					finalizeTx(&tx, false)
-					fillTimeLazy(&tx)
-					if f != nil && !f(tx) {
-						return nil
-					}
-				}
-				currentGtid = ev.Data
-				tx = Transaction{
-					GTID:      ev.Data,
-					StartPos:  startPos,
-					Timestamp: int64(h.Timestamp),
-					Txs:       make([]TxDetail, 0, 8),
-					sqlOrigin: make([]string, 0, 4),
-				}
-			case "":
-				tx.EndPos = int(h.LogPos)
-			case "rowsquery":
-				tx.EndPos = int(h.LogPos)
-				tx.sqlOrigin = append(tx.sqlOrigin, ev.Data)
-			default:
-				tx.EndPos = int(h.LogPos)
-				status := STATUS_PREPARE
-				if ev.Type == "query" {
-					switch strings.ToLower(ev.Data) {
-					case "begin":
-						if tx.TxStartTime == 0 {
-							tx.TxStartTime = int64(h.Timestamp)
-						}
-						status = STATUS_BEGIN
-					case "commit":
-						status = STATUS_COMMIT
-						tx.TxEndTime = int64(h.Timestamp)
-					case "rollback":
-						status = STATUS_ROLLBACK
-						tx.TxEndTime = int64(h.Timestamp)
-					}
-					tx.Status = status
-				}
-				if ev.DB != "" && ev.TB != "" {
-					tx.dmlEventCount++
-				}
-				tx.Txs = append(tx.Txs, TxDetail{
-					StartPos:        startPos,
-					EndPos:          int(h.LogPos),
-					Db:              ev.DB,
-					Table:           ev.TB,
-					Sql:             ev.Data,
-					SqlType:         ev.Type,
-					Rows:            ev.Rows,
-					RowCount:        int(ev.RowCnt),
-					Timestamp:       int64(h.Timestamp),
-					CompressionType: ev.CompressionType,
-				})
-			}
-		}
-	}
-}
-
-func ParseBinlogEvent(ev *replication.BinlogEvent) []BinlogEvent {
-	var res []BinlogEvent
-	var sig BinlogEvent
-
-	switch ev.Header.EventType {
-	case replication.ANONYMOUS_GTID_EVENT:
-		sig.Data = "anonymous-gtid-event:1"
-		sig.Type = "gtid"
-
-	case replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2:
-		wrEvent, ok := ev.Event.(*replication.RowsEvent)
-		if !ok {
-			return res
-		}
-		sig.DB = string(wrEvent.Table.Schema)
-		sig.TB = string(wrEvent.Table.Table)
-		sig.Type = "insert"
-		sig.RowCnt = uint32(len(wrEvent.Rows))
-		sig.Rows = wrEvent.Rows
-
-	case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
-		wrEvent, ok := ev.Event.(*replication.RowsEvent)
-		if !ok {
-			return res
-		}
-		sig.DB = string(wrEvent.Table.Schema)
-		sig.TB = string(wrEvent.Table.Table)
-		sig.Type = "update"
-		sig.RowCnt = uint32(len(wrEvent.Rows)) / 2
-		sig.Rows = wrEvent.Rows
-
-	case replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2:
-		wrEvent, ok := ev.Event.(*replication.RowsEvent)
-		if !ok {
-			return res
-		}
-		sig.DB = string(wrEvent.Table.Schema)
-		sig.TB = string(wrEvent.Table.Table)
-		sig.Type = "delete"
-		sig.RowCnt = uint32(len(wrEvent.Rows))
-		sig.Rows = wrEvent.Rows
-
-	case replication.ROWS_QUERY_EVENT:
-		queryEvent, ok := ev.Event.(*replication.RowsQueryEvent)
-		if !ok {
-			return res
-		}
-		sig.Data = string(queryEvent.Query)
-		sig.Type = "rowsquery"
-
-	case replication.QUERY_EVENT:
-		queryEvent, ok := ev.Event.(*replication.QueryEvent)
-		if !ok {
-			return res
-		}
-		sig.DB = string(queryEvent.Schema)
-		sig.Data = string(queryEvent.Query)
-		sig.Type = "query"
-
-	case replication.MARIADB_GTID_EVENT:
-		sig.Data = "begin"
-		sig.Type = "query"
-
-	case replication.XID_EVENT:
-		sig.Data = "commit"
-		sig.Type = "query"
-
-	case replication.GTID_EVENT:
-		ge, ok := ev.Event.(*replication.GTIDEvent)
-		if !ok {
-			return res
-		}
-		gid, err := gtid.Parse(fmt.Sprintf("%s:%d", bytesToUuid(ge.SID), ge.GNO))
-		if err != nil {
-			sig.Data = fmt.Sprintf("invalid-gtid:%s:%d", bytesToUuid(ge.SID), ge.GNO)
-		} else {
-			sig.Data = gid.String()
-		}
-		sig.Type = "gtid"
-
-	case replication.TRANSACTION_PAYLOAD_EVENT:
-		ge, ok := ev.Event.(*replication.TransactionPayloadEvent)
-		if !ok {
-			return res
-		}
-		for _, val := range ge.Events {
-			res = append(res, ParseBinlogEvent(val)...)
-		}
-		compressionType := getCompressionTypeName(ge.CompressionType)
-		for idx := range res {
-			res[idx].CompressionType = compressionType
-		}
-		return res
-	}
-
-	res = append(res, sig)
-	return res
-}
-
-func getCompressionTypeName(code uint64) string {
-	switch code {
-	case CompressionZSTD:
-		return "ZSTD"
-	case CompressionNone:
-		return ""
-	default:
-		return fmt.Sprintf("UNKNOWN(%d)", code)
-	}
-}
-
-func ParseBinlogWithFilter(path string, pos int64, filter BinlogFilter, fx func(Transaction) bool) error {
-	if !staros.Exists(path) {
-		return os.ErrNotExist
-	}
-
-	f, err := os.Open(path)
-	if err != nil {
-		return err
-	}
-	defer f.Close()
-
-	parser := replication.NewBinlogParser()
-	parser.SetParseTime(false)
-	parser.SetUseDecimal(false)
-
-	if pos != 0 {
-		if err := seekToPosition(f, parser, pos); err != nil {
-			return err
-		}
-	} else {
-		if err := validateBinlogHeader(f); err != nil {
-			return err
-		}
-	}
-
-	br := bufio.NewReaderSize(f, defaultReadBufSize)
-	return parseBinlogWithFilter(br, parser, filter, fx)
-}
-
-func seekToPosition(f *os.File, parser *replication.BinlogParser, pos int64) error {
-	if err := validateBinlogHeader(f); err != nil {
-		return err
-	}
-
-	headBuf := make([]byte, replication.EventHeaderSize)
-	for {
-		h, err := readEventHeader(f, parser, headBuf)
-		if err != nil {
-			return fmt.Errorf("seek to position failed: %w", err)
-		}
-		body, err := readEventBody(f, h)
-		if err != nil {
-			return err
-		}
-		_, err = parseEvent(parser, h, body)
-		putBodyBuf(body)
-		if err != nil {
-			return err
-		}
-		if h.EventType == replication.FORMAT_DESCRIPTION_EVENT || h.EventType == replication.GTID_EVENT {
-			break
-		}
-	}
-
-	if _, err := f.Seek(pos, io.SeekStart); err != nil {
-		return fmt.Errorf("seek to pos %d failed: %w", pos, err)
-	}
-	return nil
-}
-
-func parseBinlogWithFilter(r io.Reader, parser *replication.BinlogParser, filter BinlogFilter, fn func(Transaction) bool) error {
-	var subGtid, inGtid, exGtid *gtid.Gtid
-	var err error
-
-	includeMatcher, excludeMatcher := prepareTableMatchers(filter)
-
-	if filter.IncludeGtid != "" {
-		inGtid, err = gtid.Parse(filter.IncludeGtid)
-		if err != nil {
-			return fmt.Errorf("parse include gtid failed: %w", err)
-		}
-		subGtid = inGtid.Clone()
-	}
-	if filter.ExcludeGtid != "" {
-		exGtid, err = gtid.Parse(filter.ExcludeGtid)
-		if err != nil {
-			return fmt.Errorf("parse exclude gtid failed: %w", err)
-		}
-	}
-
-	var (
-		tbMapPos       uint32
-		skipCurrentTxn bool
-		tx             Transaction
-		headBuf        = make([]byte, replication.EventHeaderSize)
-	)
-	currentGtid := ""
-
-	callFn := func(tx Transaction) bool {
-		if fn == nil {
-			return true
-		}
-
-		fillTimeLazy(&tx)
-
-		if !filter.StartDate.IsZero() && filter.StartDate.After(tx.Time) {
-			return true
-		}
-		if !filter.EndDate.IsZero() && filter.EndDate.Before(tx.Time) {
-			return true
-		}
-		if filter.StartPos != 0 && filter.StartPos > tx.StartPos {
-			return true
-		}
-		if filter.EndPos != 0 && filter.EndPos < tx.EndPos {
-			return true
-		}
-		if filter.BigThan != 0 && filter.BigThan > tx.Size {
-			return true
-		}
-		if filter.SmallThan != 0 && filter.SmallThan < tx.Size {
-			return true
-		}
-		if !filter.OnlyShowGtid && filter.OnlyShowDML && tx.dmlEventCount == 0 {
-			return true
-		}
-
-		var txs []TxDetail
-		var matched bool
-
-		for _, t := range tx.Txs {
-			includeMatch := includeMatcher != nil && includeMatcher.match(t.Db, t.Table)
-			excludeMatch := excludeMatcher != nil && excludeMatcher.match(t.Db, t.Table)
-
-			if t.Db == "" && t.Table == "" {
-				if includeMatcher != nil && !filter.IncludeBlank {
-					continue
-				}
-				if excludeMatcher != nil && filter.ExcludeBlank {
-					matched = true
-					if filter.PickTxAllIfMatch {
-						return true
-					}
-					continue
-				}
-			}
-
-			if includeMatcher != nil {
-				if includeMatch {
-					matched = true
-					if filter.PickTxAllIfMatch {
-						return fn(tx)
-					}
-					txs = append(txs, t)
-				}
-			} else if excludeMatcher != nil {
-				if excludeMatch {
-					matched = true
-					if filter.PickTxAllIfMatch {
-						return true
-					}
-				} else {
-					txs = append(txs, t)
-				}
-			} else {
-				txs = append(txs, t)
-			}
-		}
-
-		if matched {
-			tx.Txs = txs
-		}
-		if !matched && includeMatcher != nil {
-			return true
-		}
-		if len(tx.Txs) == 0 && matched {
-			return true
-		}
-		return fn(tx)
-	}
-
-	for {
-		h, err := readEventHeader(r, parser, headBuf)
-		if err == io.EOF {
-			if !tx.Time.IsZero() || tx.Timestamp != 0 {
-				finalizeTx(&tx, filter.OnlyShowGtid)
-				callFn(tx)
-			}
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-
-		// GTID-only fast path
-		if filter.OnlyShowGtid {
-			if h.EventType != replication.GTID_EVENT && h.EventType != replication.ANONYMOUS_GTID_EVENT {
-				if err := skipEventBody(r, h); err != nil {
-					return err
-				}
-				continue
-			}
-
-			body, err := readEventBody(r, h)
-			if err != nil {
-				return err
-			}
-			e, err := parseEvent(parser, h, body)
-			putBodyBuf(body)
-			if err != nil {
-				return err
-			}
-
-			evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
-			for _, ev := range evs {
-				if ev.Type != "gtid" {
-					continue
-				}
-				startPos := int(h.LogPos - h.EventSize)
-
-				if filter.EndPos != 0 && startPos > filter.EndPos {
-					continue
-				}
-				if filter.StartPos != 0 && startPos < filter.StartPos {
-					continue
-				}
-
-				if currentGtid != "" {
-					tx.EndPos = startPos - 1
-					finalizeTx(&tx, true)
-					if !callFn(tx) {
-						return nil
-					}
-					if subGtid != nil {
-						if err := subGtid.Sub(tx.GTID); err == nil && subGtid.EventCount() == 0 {
-							return nil
-						}
-					}
-					tx = Transaction{}
-				}
-
-				currentGtid = ev.Data
-
-				if inGtid != nil {
-					if c, _ := inGtid.Contain(ev.Data); !c {
-						tx = Transaction{}
-						currentGtid = ""
-						continue
-					}
-				}
-				if exGtid != nil {
-					if c, _ := exGtid.Contain(ev.Data); c {
-						currentGtid = ""
-						tx = Transaction{}
-						continue
-					}
-				}
-
-				tx = Transaction{
-					GTID:      ev.Data,
-					StartPos:  startPos,
-					EndPos:    startPos,
-					Timestamp: int64(h.Timestamp),
-				}
-			}
-			continue
-		}
-
-		// 先处理GTID事件（决定当前事务是否命中）
-		if h.EventType == replication.GTID_EVENT || h.EventType == replication.ANONYMOUS_GTID_EVENT {
-			body, err := readEventBody(r, h)
-			if err != nil {
-				return err
-			}
-			e, err := parseEvent(parser, h, body)
-			putBodyBuf(body)
-			if err != nil {
-				return err
-			}
-
-			evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
-			for _, ev := range evs {
-				if ev.Type != "gtid" {
-					continue
-				}
-				startPos := int(h.LogPos - h.EventSize)
-
-				if currentGtid != "" {
-					finalizeTx(&tx, false)
-					if !callFn(tx) {
-						return nil
-					}
-					if subGtid != nil {
-						if err := subGtid.Sub(tx.GTID); err == nil && subGtid.EventCount() == 0 {
-							return nil
-						}
-					}
-					tx = Transaction{}
-				}
-
-				currentGtid = ev.Data
-				skipCurrentTxn = false
-
-				if filter.EndPos != 0 && startPos > filter.EndPos {
-					skipCurrentTxn = true
-				}
-				if filter.StartPos != 0 && startPos < filter.StartPos {
-					skipCurrentTxn = true
-				}
-				if inGtid != nil {
-					if c, _ := inGtid.Contain(ev.Data); !c {
-						skipCurrentTxn = true
-					}
-				}
-				if exGtid != nil {
-					if c, _ := exGtid.Contain(ev.Data); c {
-						skipCurrentTxn = true
-					}
-				}
-
-				if !skipCurrentTxn {
-					tx = Transaction{
-						GTID:      ev.Data,
-						StartPos:  startPos,
-						Timestamp: int64(h.Timestamp),
-						Txs:       make([]TxDetail, 0, 8),
-						sqlOrigin: make([]string, 0, 4),
-					}
-				} else {
-					tx = Transaction{}
-				}
-			}
-			continue
-		}
-
-		// 未命中事务：零解析到底
-		if skipCurrentTxn {
-			if err := skipEventBody(r, h); err != nil {
-				return err
-			}
-			continue
-		}
-
-		body, err := readEventBody(r, h)
-		if err != nil {
-			return err
-		}
-		e, err := parseEvent(parser, h, body)
-		putBodyBuf(body)
-		if err != nil {
-			return err
-		}
-
-		if h.EventType == replication.TABLE_MAP_EVENT {
-			tbMapPos = h.LogPos - h.EventSize
-		}
-
-		evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
-		for _, ev := range evs {
-			startPos := 0
-			if ev.Type == "query" || ev.Type == "gtid" {
-				startPos = int(h.LogPos - h.EventSize)
-			} else {
-				startPos = int(tbMapPos)
-			}
-
-			switch ev.Type {
-			case "":
-				tx.EndPos = int(h.LogPos)
-
-			case "rowsquery":
-				tx.EndPos = int(h.LogPos)
-				tx.sqlOrigin = append(tx.sqlOrigin, ev.Data)
-
-			default:
-				tx.EndPos = int(h.LogPos)
-				status := STATUS_PREPARE
-				if ev.Type == "query" {
-					switch strings.ToLower(ev.Data) {
-					case "begin":
-						if tx.TxStartTime == 0 {
-							tx.TxStartTime = int64(h.Timestamp)
-						}
-						status = STATUS_BEGIN
-					case "commit":
-						status = STATUS_COMMIT
-						tx.TxEndTime = int64(h.Timestamp)
-					case "rollback":
-						status = STATUS_ROLLBACK
-						tx.TxEndTime = int64(h.Timestamp)
-					}
-					tx.Status = status
-				}
-				if ev.DB != "" && ev.TB != "" {
-					tx.dmlEventCount++
-				}
-				tx.Txs = append(tx.Txs, TxDetail{
-					StartPos:        startPos,
-					EndPos:          int(h.LogPos),
-					Db:              ev.DB,
-					Table:           ev.TB,
-					Sql:             ev.Data,
-					SqlType:         ev.Type,
-					Rows:            ev.Rows,
-					RowCount:        int(ev.RowCnt),
-					Timestamp:       int64(h.Timestamp),
-					CompressionType: ev.CompressionType,
-				})
-			}
-		}
-	}
-}
-
-func prepareTableMatchers(filter BinlogFilter) (includeMatcher, excludeMatcher *tableMatcher) {
-	if len(filter.IncludeTables) > 0 {
-		includeMatcher = buildTableMatcher(filter.IncludeTables)
-	}
-	if len(filter.ExcludeTables) > 0 {
-		excludeMatcher = buildTableMatcher(filter.ExcludeTables)
-	}
-	return includeMatcher, excludeMatcher
-}
-
-func buildTableMatcher(patterns []string) *tableMatcher {
-	m := &tableMatcher{
-		exactMatch: make(map[string]bool),
-		dbWildcard: make(map[string]bool),
-		tbWildcard: make(map[string]bool),
-	}
-
-	for _, pattern := range patterns {
-		if pattern == "*.*" {
-			m.matchAll = true
-			continue
-		}
-		parts := strings.Split(pattern, ".")
-		if len(parts) != 2 {
-			continue
-		}
-		db, tb := parts[0], parts[1]
-		if db == "*" && tb == "*" {
-			m.matchAll = true
-		} else if db == "*" {
-			m.tbWildcard[tb] = true
-		} else if tb == "*" {
-			m.dbWildcard[db] = true
-		} else {
-			m.exactMatch[pattern] = true
-		}
-	}
-	return m
-}
+// Parsing implementation is split across:
+// - parse_types.go
+// - parse_io.go
+// - parse_event_convert.go
+// - parse_stream.go
+// - parse_filter.go
@@ -0,0 +1,288 @@
+package binlog
+
+import (
+	"fmt"
+
+	"b612.me/mysql/gtid"
+	"github.com/starainrt/go-mysql/mysql"
+	"github.com/starainrt/go-mysql/replication"
+)
+
+func ParseBinlogEvent(ev *replication.BinlogEvent) []BinlogEvent {
+	if ev == nil || ev.Header == nil {
+		return nil
+	}
+
+	var buf [1]BinlogEvent
+	sig := &buf[0]
+	fillEventHeader(sig, ev)
+
+	switch ev.Header.EventType {
+	case replication.ANONYMOUS_GTID_EVENT:
+		sig.Data = "anonymous-gtid-event:1"
+		sig.Type = "gtid"
+		if ge, ok := ev.Event.(*replication.GTIDEvent); ok {
+			fillGTIDMetadata(sig, ge)
+		}
+
+	case replication.WRITE_ROWS_EVENTv1, replication.WRITE_ROWS_EVENTv2:
+		wrEvent, ok := ev.Event.(*replication.RowsEvent)
+		if !ok {
+			return nil
+		}
+		sig.DB = string(wrEvent.Table.Schema)
+		sig.TB = string(wrEvent.Table.Table)
+		sig.Type = "insert"
+		sig.RowCnt = uint32(len(wrEvent.Rows))
+		sig.Rows = normalizeRowsByUnsigned(wrEvent)
+		sig.ColumnTypes = cloneColumnTypes(wrEvent.Table)
+		sig.ColumnCollationIDs = buildColumnCollationIDs(wrEvent.Table)
+
+	case replication.UPDATE_ROWS_EVENTv1, replication.UPDATE_ROWS_EVENTv2:
+		wrEvent, ok := ev.Event.(*replication.RowsEvent)
+		if !ok {
+			return nil
+		}
+		sig.DB = string(wrEvent.Table.Schema)
+		sig.TB = string(wrEvent.Table.Table)
+		sig.Type = "update"
+		sig.RowCnt = uint32(len(wrEvent.Rows)) / 2
+		sig.Rows = normalizeRowsByUnsigned(wrEvent)
+		sig.ColumnTypes = cloneColumnTypes(wrEvent.Table)
+		sig.ColumnCollationIDs = buildColumnCollationIDs(wrEvent.Table)
+
+	case replication.DELETE_ROWS_EVENTv1, replication.DELETE_ROWS_EVENTv2:
+		wrEvent, ok := ev.Event.(*replication.RowsEvent)
+		if !ok {
+			return nil
+		}
+		sig.DB = string(wrEvent.Table.Schema)
+		sig.TB = string(wrEvent.Table.Table)
+		sig.Type = "delete"
+		sig.RowCnt = uint32(len(wrEvent.Rows))
+		sig.Rows = normalizeRowsByUnsigned(wrEvent)
+		sig.ColumnTypes = cloneColumnTypes(wrEvent.Table)
+		sig.ColumnCollationIDs = buildColumnCollationIDs(wrEvent.Table)
+
+	case replication.TABLE_MAP_EVENT:
+		tableEvent, ok := ev.Event.(*replication.TableMapEvent)
+		if !ok {
+			return nil
+		}
+		sig.DB = string(tableEvent.Schema)
+		sig.TB = string(tableEvent.Table)
+		sig.Type = "tablemap"
+
+	case replication.ROWS_QUERY_EVENT:
+		queryEvent, ok := ev.Event.(*replication.RowsQueryEvent)
+		if !ok {
+			return nil
+		}
+		sig.Data = string(queryEvent.Query)
+		sig.Type = "rowsquery"
+
+	case replication.QUERY_EVENT:
+		queryEvent, ok := ev.Event.(*replication.QueryEvent)
+		if !ok {
+			return nil
+		}
+		sig.DB = string(queryEvent.Schema)
+		sig.Data = string(queryEvent.Query)
+		sig.Type = "query"
+
+	case replication.MARIADB_GTID_EVENT:
+		sig.Data = "begin"
+		sig.Type = "query"
+
+	case replication.XID_EVENT:
+		sig.Data = "commit"
+		sig.Type = "query"
+
+	case replication.GTID_EVENT:
+		ge, ok := ev.Event.(*replication.GTIDEvent)
+		if !ok {
+			return nil
+		}
+		gid, err := gtid.Parse(fmt.Sprintf("%s:%d", bytesToUuid(ge.SID), ge.GNO))
+		if err != nil {
+			sig.Data = fmt.Sprintf("invalid-gtid:%s:%d", bytesToUuid(ge.SID), ge.GNO)
+		} else {
+			sig.Data = gid.String()
+		}
+		sig.Type = "gtid"
+		fillGTIDMetadata(sig, ge)
+
+	case replication.TRANSACTION_PAYLOAD_EVENT:
+		ge, ok := ev.Event.(*replication.TransactionPayloadEvent)
+		if !ok {
+			return nil
+		}
+		res := make([]BinlogEvent, 0, len(ge.Events))
+		for _, val := range ge.Events {
+			res = append(res, ParseBinlogEvent(val)...)
+		}
+		compressionType := getCompressionTypeName(ge.CompressionType)
+		for idx := range res {
+			res[idx].CompressionType = compressionType
+		}
+		return res
+	}
+
+	// 返回栈上数组的切片。调用方在当前迭代内立即消费，不持有跨迭代引用，安全。
+	return buf[:]
+}
+
+func fillEventHeader(sig *BinlogEvent, ev *replication.BinlogEvent) {
+	if sig == nil || ev == nil || ev.Header == nil {
+		return
+	}
+	sig.EventType = byte(ev.Header.EventType)
+	sig.ServerID = ev.Header.ServerID
+	sig.Timestamp = ev.Header.Timestamp
+	sig.LogPos = ev.Header.LogPos
+	sig.EventSize = ev.Header.EventSize
+}
+
+func fillGTIDMetadata(sig *BinlogEvent, ge *replication.GTIDEvent) {
+	if sig == nil || ge == nil {
+		return
+	}
+	sig.LastCommitted = ge.LastCommitted
+	sig.SequenceNumber = ge.SequenceNumber
+	sig.TransactionLength = ge.TransactionLength
+	sig.ImmediateCommitTimestamp = ge.ImmediateCommitTimestamp
+	sig.OriginalCommitTimestamp = ge.OriginalCommitTimestamp
+}
+
+func normalizeRowsByUnsigned(wrEvent *replication.RowsEvent) [][]interface{} {
+	if wrEvent == nil || wrEvent.Table == nil || len(wrEvent.Rows) == 0 {
+		if wrEvent == nil {
+			return nil
+		}
+		return wrEvent.Rows
+	}
+
+	unsignedMap := wrEvent.Table.UnsignedMap()
+	if len(unsignedMap) == 0 {
+		return wrEvent.Rows
+	}
+
+	columnTypes := wrEvent.Table.ColumnType
+	if len(columnTypes) == 0 {
+		return wrEvent.Rows
+	}
+
+	for rowIdx := range wrEvent.Rows {
+		row := wrEvent.Rows[rowIdx]
+		for colIdx := range row {
+			if !unsignedMap[colIdx] {
+				continue
+			}
+			if colIdx >= len(columnTypes) {
+				continue
+			}
+			row[colIdx] = normalizeUnsignedValue(row[colIdx], columnTypes[colIdx])
+		}
+	}
+	return wrEvent.Rows
+}
+
+func cloneColumnTypes(table *replication.TableMapEvent) []int {
+	if table == nil || len(table.ColumnType) == 0 {
+		return nil
+	}
+	ret := make([]int, len(table.ColumnType))
+	for i, t := range table.ColumnType {
+		ret[i] = int(t)
+	}
+	return ret
+}
+
+func buildColumnCollationIDs(table *replication.TableMapEvent) []uint64 {
+	if table == nil {
+		return nil
+	}
+
+	columnCount := len(table.ColumnType)
+	if columnCount == 0 && table.ColumnCount > 0 {
+		columnCount = int(table.ColumnCount)
+	}
+	if columnCount == 0 {
+		return nil
+	}
+
+	ret := make([]uint64, columnCount)
+	hasValue := false
+
+	for idx, collationID := range table.CollationMap() {
+		if idx < 0 || idx >= columnCount {
+			continue
+		}
+		ret[idx] = collationID
+		hasValue = hasValue || collationID != 0
+	}
+	for idx, collationID := range table.EnumSetCollationMap() {
+		if idx < 0 || idx >= columnCount {
+			continue
+		}
+		if ret[idx] == 0 {
+			ret[idx] = collationID
+		}
+		hasValue = hasValue || collationID != 0
+	}
+
+	if !hasValue {
+		return nil
+	}
+	return ret
+}
+
+func normalizeUnsignedValue(v interface{}, colType byte) interface{} {
+	signed, ok := signedToInt64(v)
+	if !ok {
+		return v
+	}
+
+	switch colType {
+	case mysql.MYSQL_TYPE_TINY:
+		return uint8(signed)
+	case mysql.MYSQL_TYPE_SHORT:
+		return uint16(signed)
+	case mysql.MYSQL_TYPE_INT24:
+		return uint32(uint32(int32(signed)) & 0x00FFFFFF)
+	case mysql.MYSQL_TYPE_LONG:
+		return uint32(signed)
+	case mysql.MYSQL_TYPE_LONGLONG:
+		return uint64(signed)
+	default:
+		return v
+	}
+}
+
+func signedToInt64(v interface{}) (int64, bool) {
+	switch x := v.(type) {
+	case int8:
+		return int64(x), true
+	case int16:
+		return int64(x), true
+	case int32:
+		return int64(x), true
+	case int64:
+		return x, true
+	case int:
+		return int64(x), true
+	default:
+		return 0, false
+	}
+}
+
+func getCompressionTypeName(code uint64) string {
+	switch code {
+	case CompressionZSTD:
+		return "ZSTD"
+	case CompressionNone:
+		return ""
+	default:
+		return fmt.Sprintf("UNKNOWN(%d)", code)
+	}
+}
@@ -0,0 +1,121 @@
+package binlog
+
+import (
+	"testing"
+
+	"github.com/starainrt/go-mysql/mysql"
+	"github.com/starainrt/go-mysql/replication"
+)
+
+func TestParseBinlogEvent_TableMapEvent(t *testing.T) {
+	ev := &replication.BinlogEvent{
+		Header: &replication.EventHeader{EventType: replication.TABLE_MAP_EVENT},
+		Event: &replication.TableMapEvent{
+			Schema: []byte("db1"),
+			Table:  []byte("tb1"),
+		},
+	}
+
+	events := ParseBinlogEvent(ev)
+	if len(events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(events))
+	}
+	if events[0].Type != "tablemap" {
+		t.Fatalf("expected tablemap event type, got %q", events[0].Type)
+	}
+	if events[0].DB != "db1" || events[0].TB != "tb1" {
+		t.Fatalf("unexpected db/table: %s.%s", events[0].DB, events[0].TB)
+	}
+}
+
+func TestParseBinlogEvent_TransactionPayloadContainsTableMap(t *testing.T) {
+	table := &replication.TableMapEvent{
+		Schema:     []byte("db2"),
+		Table:      []byte("tb2"),
+		ColumnType: []byte{mysql.MYSQL_TYPE_LONG},
+	}
+
+	payload := &replication.TransactionPayloadEvent{
+		CompressionType: CompressionZSTD,
+		Events: []*replication.BinlogEvent{
+			{
+				Header: &replication.EventHeader{EventType: replication.TABLE_MAP_EVENT},
+				Event:  table,
+			},
+			{
+				Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2},
+				Event: &replication.RowsEvent{
+					Table: table,
+					Rows:  [][]interface{}{{int32(1)}},
+				},
+			},
+		},
+	}
+
+	ev := &replication.BinlogEvent{
+		Header: &replication.EventHeader{EventType: replication.TRANSACTION_PAYLOAD_EVENT},
+		Event:  payload,
+	}
+
+	events := ParseBinlogEvent(ev)
+	if len(events) != 2 {
+		t.Fatalf("expected 2 events from payload, got %d", len(events))
+	}
+	if events[0].Type != "tablemap" {
+		t.Fatalf("expected first payload event to be tablemap, got %q", events[0].Type)
+	}
+	if events[1].Type != "insert" {
+		t.Fatalf("expected second payload event to be insert, got %q", events[1].Type)
+	}
+	if events[0].CompressionType != "ZSTD" || events[1].CompressionType != "ZSTD" {
+		t.Fatalf("expected payload events to carry compression type, got %q/%q", events[0].CompressionType, events[1].CompressionType)
+	}
+}
+
+func TestParseBinlogEvent_NilInput(t *testing.T) {
+	if got := ParseBinlogEvent(nil); got != nil {
+		t.Fatalf("expected nil for nil event, got %#v", got)
+	}
+	if got := ParseBinlogEvent(&replication.BinlogEvent{}); got != nil {
+		t.Fatalf("expected nil for missing header, got %#v", got)
+	}
+}
+
+func TestParseBinlogEvent_GTIDMetadata(t *testing.T) {
+	ev := &replication.BinlogEvent{
+		Header: &replication.EventHeader{
+			EventType: replication.GTID_EVENT,
+			ServerID:  12,
+			Timestamp: 123456,
+			LogPos:    456,
+			EventSize: 64,
+		},
+		Event: &replication.GTIDEvent{
+			SID:                      []byte{0x74, 0xde, 0xc5, 0xa0, 0x3a, 0xc7, 0x11, 0xf0, 0xba, 0x0c, 0xfa, 0x16, 0x3e, 0xea, 0x29, 0x9f},
+			GNO:                      42,
+			LastCommitted:            40,
+			SequenceNumber:           42,
+			TransactionLength:        2048,
+			ImmediateCommitTimestamp: 1700000000000001,
+			OriginalCommitTimestamp:  1700000000000000,
+		},
+	}
+
+	events := ParseBinlogEvent(ev)
+	if len(events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(events))
+	}
+	got := events[0]
+	if got.Type != "gtid" {
+		t.Fatalf("expected gtid event, got %q", got.Type)
+	}
+	if got.LastCommitted != 40 || got.SequenceNumber != 42 || got.TransactionLength != 2048 {
+		t.Fatalf("logical metadata mismatch: last=%d seq=%d len=%d", got.LastCommitted, got.SequenceNumber, got.TransactionLength)
+	}
+	if got.ImmediateCommitTimestamp != 1700000000000001 || got.OriginalCommitTimestamp != 1700000000000000 {
+		t.Fatalf("commit timestamps mismatch: immediate=%d original=%d", got.ImmediateCommitTimestamp, got.OriginalCommitTimestamp)
+	}
+	if got.ServerID != 12 || got.Timestamp != 123456 || got.LogPos != 456 || got.EventSize != 64 {
+		t.Fatalf("header metadata mismatch: %#v", got)
+	}
+}
@@ -0,0 +1,95 @@
+package binlog
+
+import (
+	"testing"
+
+	"github.com/starainrt/go-mysql/mysql"
+	"github.com/starainrt/go-mysql/replication"
+)
+
+func TestNormalizeRowsByUnsigned_AllIntegerKinds(t *testing.T) {
+	event := &replication.RowsEvent{
+		Table: &replication.TableMapEvent{
+			ColumnCount:      5,
+			ColumnType:       []byte{mysql.MYSQL_TYPE_TINY, mysql.MYSQL_TYPE_SHORT, mysql.MYSQL_TYPE_INT24, mysql.MYSQL_TYPE_LONG, mysql.MYSQL_TYPE_LONGLONG},
+			SignednessBitmap: []byte{0xF8},
+		},
+		Rows: [][]interface{}{{int8(-1), int16(-2), int32(-1), int32(-1), int64(-1)}},
+	}
+
+	got := normalizeRowsByUnsigned(event)
+	row := got[0]
+
+	if v, ok := row[0].(uint8); !ok || v != 255 {
+		t.Fatalf("tiny unsigned mismatch: %T %v", row[0], row[0])
+	}
+	if v, ok := row[1].(uint16); !ok || v != 65534 {
+		t.Fatalf("short unsigned mismatch: %T %v", row[1], row[1])
+	}
+	if v, ok := row[2].(uint32); !ok || v != 16777215 {
+		t.Fatalf("int24 unsigned mismatch: %T %v", row[2], row[2])
+	}
+	if v, ok := row[3].(uint32); !ok || v != 4294967295 {
+		t.Fatalf("long unsigned mismatch: %T %v", row[3], row[3])
+	}
+	if v, ok := row[4].(uint64); !ok || v != 18446744073709551615 {
+		t.Fatalf("longlong unsigned mismatch: %T %v", row[4], row[4])
+	}
+}
+
+func TestNormalizeRowsByUnsigned_NoSignednessMetadata(t *testing.T) {
+	event := &replication.RowsEvent{
+		Table: &replication.TableMapEvent{
+			ColumnCount: 1,
+			ColumnType:  []byte{mysql.MYSQL_TYPE_LONGLONG},
+		},
+		Rows: [][]interface{}{{int64(-1)}},
+	}
+
+	got := normalizeRowsByUnsigned(event)
+	if v, ok := got[0][0].(int64); !ok || v != -1 {
+		t.Fatalf("value should remain signed when metadata missing: %T %v", got[0][0], got[0][0])
+	}
+}
+
+func TestParseBinlogEvent_IncludeColumnMetadata(t *testing.T) {
+	event := &replication.RowsEvent{
+		Table: &replication.TableMapEvent{
+			ColumnCount:    3,
+			ColumnType:     []byte{mysql.MYSQL_TYPE_VAR_STRING, mysql.MYSQL_TYPE_LONG, mysql.MYSQL_TYPE_VAR_STRING},
+			DefaultCharset: []uint64{45}, // utf8mb4_general_ci
+		},
+		Rows: [][]interface{}{{"name", int32(1), "desc"}},
+	}
+
+	ev := &replication.BinlogEvent{
+		Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2},
+		Event:  event,
+	}
+
+	events := ParseBinlogEvent(ev)
+	if len(events) != 1 {
+		t.Fatalf("expected 1 event, got %d", len(events))
+	}
+	got := events[0]
+
+	if len(got.ColumnTypes) != 3 {
+		t.Fatalf("unexpected column type length: %d", len(got.ColumnTypes))
+	}
+	if got.ColumnTypes[0] != int(mysql.MYSQL_TYPE_VAR_STRING) || got.ColumnTypes[1] != int(mysql.MYSQL_TYPE_LONG) {
+		t.Fatalf("unexpected column types: %v", got.ColumnTypes)
+	}
+
+	if len(got.ColumnCollationIDs) != 3 {
+		t.Fatalf("unexpected column collation length: %d", len(got.ColumnCollationIDs))
+	}
+	if got.ColumnCollationIDs[0] != 45 {
+		t.Fatalf("unexpected collation for column 0: %d", got.ColumnCollationIDs[0])
+	}
+	if got.ColumnCollationIDs[2] != 45 {
+		t.Fatalf("unexpected collation for column 2: %d", got.ColumnCollationIDs[2])
+	}
+	if got.ColumnCollationIDs[1] != 0 {
+		t.Fatalf("non-character column should keep zero collation: %d", got.ColumnCollationIDs[1])
+	}
+}
@@ -0,0 +1,692 @@
+package binlog
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"b612.me/mysql/gtid"
+	"b612.me/staros"
+	"github.com/starainrt/go-mysql/replication"
+)
+
+func ParseBinlogWithFilter(path string, pos int64, filter BinlogFilter, fx func(Transaction) bool) error {
+	return ParseBinlogWithOptions(path, ParseOptions{
+		StartPos: pos,
+		Filter:   filter,
+	}, fx)
+}
+
+func ParseBinlogWithOptions(path string, opts ParseOptions, fx func(Transaction) bool) error {
+	filter := opts.Filter
+	if hasConfiguredTablePatterns(filter.IncludeTables) && hasConfiguredTablePatterns(filter.ExcludeTables) {
+		return fmt.Errorf("invalid filter: include-tables and exclude-tables cannot be set at the same time")
+	}
+
+	if !staros.Exists(path) {
+		return os.ErrNotExist
+	}
+
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	parser := replication.NewBinlogParser()
+	parser.SetParseTime(false)
+	parser.SetUseDecimal(false)
+
+	if opts.StartPos != 0 {
+		if err := seekToPosition(f, parser, opts.StartPos); err != nil {
+			return err
+		}
+	} else {
+		if err := validateBinlogHeader(f); err != nil {
+			return err
+		}
+	}
+
+	fileSize := int64(0)
+	if info, err := f.Stat(); err == nil {
+		fileSize = info.Size()
+	}
+	br := bufio.NewReaderSize(f, defaultReadBufSize)
+	return parseBinlogWithFilter(br, parser, path, fileSize, opts, fx)
+}
+
+func parseBinlogWithFilter(r io.Reader, parser *replication.BinlogParser, path string, fileSize int64, opts ParseOptions, fn func(Transaction) bool) error {
+	filter := opts.Filter
+	if hasConfiguredTablePatterns(filter.IncludeTables) && hasConfiguredTablePatterns(filter.ExcludeTables) {
+		return fmt.Errorf("invalid filter: include-tables and exclude-tables cannot be set at the same time")
+	}
+	ctx := opts.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	var subGtid, inGtid, exGtid *gtid.Gtid
+	var err error
+
+	includeMatcher, excludeMatcher, err := prepareTableMatchers(filter)
+	if err != nil {
+		return err
+	}
+
+	if filter.IncludeGtid != "" {
+		inGtid, err = gtid.Parse(filter.IncludeGtid)
+		if err != nil {
+			return fmt.Errorf("parse include gtid failed: %w", err)
+		}
+		subGtid = inGtid.Clone()
+	}
+	if filter.ExcludeGtid != "" {
+		exGtid, err = gtid.Parse(filter.ExcludeGtid)
+		if err != nil {
+			return fmt.Errorf("parse exclude gtid failed: %w", err)
+		}
+	}
+
+	var (
+		tbMapPos       uint32
+		skipCurrentTxn bool
+		tx             Transaction
+		headBuf        = make([]byte, replication.EventHeaderSize)
+	)
+	currentGtid := ""
+
+	callFn := func(tx Transaction) bool {
+		if fn == nil {
+			return true
+		}
+
+		fillTimeLazy(&tx)
+
+		if !filter.StartDate.IsZero() && filter.StartDate.After(tx.Time) {
+			return true
+		}
+		if !filter.EndDate.IsZero() && filter.EndDate.Before(tx.Time) {
+			return true
+		}
+		if filter.StartPos != 0 && filter.StartPos > tx.StartPos {
+			return true
+		}
+		if filter.EndPos != 0 && filter.EndPos < tx.EndPos {
+			return true
+		}
+		if filter.BigThan != 0 && filter.BigThan > tx.Size {
+			return true
+		}
+		if filter.SmallThan != 0 && filter.SmallThan < tx.Size {
+			return true
+		}
+		if !filter.OnlyShowGtid && filter.OnlyShowDML && tx.dmlEventCount == 0 {
+			return true
+		}
+		if includeMatcher == nil && excludeMatcher == nil {
+			return fn(tx)
+		}
+
+		txs, matched, pickAll, skipAll := selectVisibleTxDetails(tx, includeMatcher, excludeMatcher, filter)
+		if pickAll {
+			return fn(tx)
+		}
+		if skipAll {
+			return true
+		}
+
+		if matched {
+			tx.Txs = txs
+			recomputeTxStatsFromVisibleDetails(&tx)
+		}
+		if !matched && includeMatcher != nil {
+			return true
+		}
+		if len(tx.Txs) == 0 && matched {
+			return true
+		}
+		return fn(tx)
+	}
+	observeProgress := func(h *replication.EventHeader, evs []BinlogEvent) bool {
+		if opts.OnProgress == nil || h == nil {
+			return true
+		}
+		eventPos := int64(h.LogPos - h.EventSize)
+		nextPos := int64(h.LogPos)
+		if len(evs) == 0 {
+			ev := BinlogEvent{
+				EventType: byte(h.EventType),
+				ServerID:  h.ServerID,
+				Timestamp: h.Timestamp,
+				LogPos:    h.LogPos,
+				EventSize: h.EventSize,
+			}
+			return opts.OnProgress(ParseProgress{
+				Path:     path,
+				Event:    ev,
+				EventPos: eventPos,
+				NextPos:  nextPos,
+				FileSize: fileSize,
+			})
+		}
+		for _, ev := range evs {
+			if !opts.OnProgress(ParseProgress{
+				Path:     path,
+				Event:    ev,
+				EventPos: eventPos,
+				NextPos:  nextPos,
+				FileSize: fileSize,
+			}) {
+				return false
+			}
+		}
+		return true
+	}
+
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		h, err := readEventHeader(r, parser, headBuf)
+		if err == io.EOF {
+			if currentGtid != "" {
+				finalizeTx(&tx, filter.OnlyShowGtid)
+				callFn(tx)
+			}
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+
+		if filter.OnlyShowGtid {
+			if h.EventType != replication.GTID_EVENT && h.EventType != replication.ANONYMOUS_GTID_EVENT {
+				if h.EventType == replication.FORMAT_DESCRIPTION_EVENT ||
+					h.EventType == replication.TABLE_MAP_EVENT {
+					body, err := readEventBody(r, h)
+					if err != nil {
+						return err
+					}
+					e, err := parseEvent(parser, h, headBuf, body)
+					if err != nil {
+						return err
+					}
+					evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
+					if !observeProgress(h, evs) {
+						return nil
+					}
+				} else {
+					if err := skipEventBody(r, h); err != nil {
+						return err
+					}
+					if !observeProgress(h, nil) {
+						return nil
+					}
+				}
+				continue
+			}
+			body, err := readEventBody(r, h)
+			if err != nil {
+				return err
+			}
+			e, err := parseEvent(parser, h, headBuf, body)
+			if err != nil {
+				return err
+			}
+
+			evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
+			if !observeProgress(h, evs) {
+				return nil
+			}
+			for _, ev := range evs {
+				if ev.Type != "gtid" {
+					continue
+				}
+				startPos := int(h.LogPos - h.EventSize)
+
+				if filter.EndPos != 0 && startPos > filter.EndPos {
+					continue
+				}
+				if filter.StartPos != 0 && startPos < filter.StartPos {
+					continue
+				}
+
+				if currentGtid != "" {
+					tx.EndPos = startPos - 1
+					finalizeTx(&tx, true)
+					if !callFn(tx) {
+						return nil
+					}
+					if subGtid != nil {
+						if err := subGtid.Sub(tx.GTID); err == nil && subGtid.EventCount() == 0 {
+							return nil
+						}
+					}
+					tx = Transaction{}
+				}
+
+				currentGtid = ev.Data
+
+				if inGtid != nil {
+					if c, _ := inGtid.Contain(ev.Data); !c {
+						tx = Transaction{}
+						currentGtid = ""
+						continue
+					}
+				}
+				if exGtid != nil {
+					if c, _ := exGtid.Contain(ev.Data); c {
+						currentGtid = ""
+						tx = Transaction{}
+						continue
+					}
+				}
+
+				tx = Transaction{
+					GTID:                     ev.Data,
+					StartPos:                 startPos,
+					EndPos:                   startPos,
+					Timestamp:                int64(h.Timestamp),
+					LastCommitted:            ev.LastCommitted,
+					SequenceNumber:           ev.SequenceNumber,
+					TransactionLength:        ev.TransactionLength,
+					ImmediateCommitTimestamp: ev.ImmediateCommitTimestamp,
+					OriginalCommitTimestamp:  ev.OriginalCommitTimestamp,
+				}
+			}
+			continue
+		}
+
+		// 先处理 GTID 事件（决定当前事务是否命中）
+		if h.EventType == replication.GTID_EVENT || h.EventType == replication.ANONYMOUS_GTID_EVENT {
+			body, err := readEventBody(r, h)
+			if err != nil {
+				return err
+			}
+			e, err := parseEvent(parser, h, headBuf, body)
+			if err != nil {
+				return err
+			}
+
+			evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
+			if !observeProgress(h, evs) {
+				return nil
+			}
+			for _, ev := range evs {
+				if ev.Type != "gtid" {
+					continue
+				}
+				startPos := int(h.LogPos - h.EventSize)
+
+				if currentGtid != "" {
+					finalizeTx(&tx, false)
+					if !callFn(tx) {
+						return nil
+					}
+					if subGtid != nil {
+						if err := subGtid.Sub(tx.GTID); err == nil && subGtid.EventCount() == 0 {
+							return nil
+						}
+					}
+					tx = Transaction{}
+				}
+
+				currentGtid = ev.Data
+				skipCurrentTxn = false
+
+				if filter.EndPos != 0 && startPos > filter.EndPos {
+					skipCurrentTxn = true
+				}
+				if filter.StartPos != 0 && startPos < filter.StartPos {
+					skipCurrentTxn = true
+				}
+				if inGtid != nil {
+					if c, _ := inGtid.Contain(ev.Data); !c {
+						skipCurrentTxn = true
+					}
+				}
+				if exGtid != nil {
+					if c, _ := exGtid.Contain(ev.Data); c {
+						skipCurrentTxn = true
+					}
+				}
+
+				if !skipCurrentTxn {
+					tx = Transaction{
+						GTID:                     ev.Data,
+						StartPos:                 startPos,
+						Timestamp:                int64(h.Timestamp),
+						LastCommitted:            ev.LastCommitted,
+						SequenceNumber:           ev.SequenceNumber,
+						TransactionLength:        ev.TransactionLength,
+						ImmediateCommitTimestamp: ev.ImmediateCommitTimestamp,
+						OriginalCommitTimestamp:  ev.OriginalCommitTimestamp,
+						Txs:                      make([]TxDetail, 0, 8),
+						sqlOrigin:                make([]string, 0, 4),
+					}
+				} else {
+					tx = Transaction{}
+					currentGtid = ""
+				}
+			}
+			continue
+		}
+
+		// 未命中事务时，TABLE_MAP_EVENT 仍需解析（parser 缓存表元数据），
+		// 其余事件可安全跳过
+		if skipCurrentTxn {
+			if h.EventType == replication.TABLE_MAP_EVENT ||
+				h.EventType == replication.FORMAT_DESCRIPTION_EVENT {
+				body, err := readEventBody(r, h)
+				if err != nil {
+					return err
+				}
+				e, err := parseEvent(parser, h, headBuf, body)
+				if err != nil {
+					return err
+				}
+				evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
+				if !observeProgress(h, evs) {
+					return nil
+				}
+			} else {
+				if err := skipEventBody(r, h); err != nil {
+					return err
+				}
+				if !observeProgress(h, nil) {
+					return nil
+				}
+			}
+			continue
+		}
+
+		body, err := readEventBody(r, h)
+		if err != nil {
+			return err
+		}
+		e, err := parseEvent(parser, h, headBuf, body)
+		if err != nil {
+			return err
+		}
+
+		if h.EventType == replication.TABLE_MAP_EVENT {
+			tbMapPos = h.LogPos - h.EventSize
+		}
+
+		evs := ParseBinlogEvent(&replication.BinlogEvent{Header: h, Event: e})
+		if !observeProgress(h, evs) {
+			return nil
+		}
+		for _, ev := range evs {
+			startPos := 0
+			if ev.Type == "query" || ev.Type == "gtid" {
+				startPos = int(h.LogPos - h.EventSize)
+			} else {
+				startPos = int(tbMapPos)
+			}
+
+			switch ev.Type {
+			case "gtid":
+				if currentGtid != "" {
+					finalizeTx(&tx, false)
+					if !callFn(tx) {
+						return nil
+					}
+					if subGtid != nil {
+						if err := subGtid.Sub(tx.GTID); err == nil && subGtid.EventCount() == 0 {
+							return nil
+						}
+					}
+				}
+				currentGtid = ev.Data
+				tx = Transaction{
+					GTID:                     ev.Data,
+					StartPos:                 startPos,
+					Timestamp:                int64(h.Timestamp),
+					LastCommitted:            ev.LastCommitted,
+					SequenceNumber:           ev.SequenceNumber,
+					TransactionLength:        ev.TransactionLength,
+					ImmediateCommitTimestamp: ev.ImmediateCommitTimestamp,
+					OriginalCommitTimestamp:  ev.OriginalCommitTimestamp,
+					Txs:                      make([]TxDetail, 0, 8),
+					sqlOrigin:                make([]string, 0, 4),
+				}
+
+			case "":
+				tx.EndPos = int(h.LogPos)
+
+			case "tablemap":
+				tx.EndPos = int(h.LogPos)
+				tbMapPos = h.LogPos - h.EventSize
+
+			case "rowsquery":
+				tx.EndPos = int(h.LogPos)
+				tx.sqlOrigin = append(tx.sqlOrigin, ev.Data)
+
+			default:
+				tx.EndPos = int(h.LogPos)
+				if ev.Type == "query" {
+					if equalFoldShort(ev.Data, "begin") {
+						if tx.TxStartTime == 0 {
+							tx.TxStartTime = int64(h.Timestamp)
+						}
+						tx.Status = STATUS_BEGIN
+					} else if equalFoldShort(ev.Data, "commit") {
+						tx.Status = STATUS_COMMIT
+						tx.TxEndTime = int64(h.Timestamp)
+					} else if equalFoldShort(ev.Data, "rollback") {
+						tx.Status = STATUS_ROLLBACK
+						tx.TxEndTime = int64(h.Timestamp)
+					}
+				}
+				if ev.DB != "" && ev.TB != "" {
+					tx.dmlEventCount++
+				}
+				tx.Txs = append(tx.Txs, TxDetail{
+					StartPos:           startPos,
+					EndPos:             int(h.LogPos),
+					Db:                 ev.DB,
+					Table:              ev.TB,
+					Sql:                ev.Data,
+					SqlType:            ev.Type,
+					Rows:               ev.Rows,
+					ColumnTypes:        ev.ColumnTypes,
+					ColumnCollationIDs: ev.ColumnCollationIDs,
+					RowCount:           int(ev.RowCnt),
+					Timestamp:          int64(h.Timestamp),
+					CompressionType:    ev.CompressionType,
+				})
+			}
+		}
+	}
+}
+
+func selectVisibleTxDetails(tx Transaction, includeMatcher, excludeMatcher *tableMatcher, filter BinlogFilter) ([]TxDetail, bool, bool, bool) {
+	txs := make([]TxDetail, 0, len(tx.Txs))
+	matched := false
+
+	for _, t := range tx.Txs {
+		includeMatch := includeMatcher != nil && includeMatcher.match(t.Db, t.Table)
+		excludeMatch := excludeMatcher != nil && excludeMatcher.match(t.Db, t.Table)
+
+		if t.Db == "" && t.Table == "" {
+			if includeMatcher != nil {
+				if filter.IncludeBlank {
+					matched = true
+					if filter.PickTxAllIfMatch {
+						return nil, true, true, false
+					}
+					txs = append(txs, t)
+				}
+				continue
+			}
+
+			if excludeMatcher != nil {
+				if filter.ExcludeBlank {
+					matched = true
+					if filter.PickTxAllIfMatch {
+						return nil, true, false, true
+					}
+					continue
+				}
+				txs = append(txs, t)
+				continue
+			}
+
+			txs = append(txs, t)
+			continue
+		}
+
+		if includeMatcher != nil {
+			if includeMatch {
+				matched = true
+				if filter.PickTxAllIfMatch {
+					return nil, true, true, false
+				}
+				txs = append(txs, t)
+			}
+			continue
+		}
+
+		if excludeMatcher != nil {
+			if excludeMatch {
+				matched = true
+				if filter.PickTxAllIfMatch {
+					return nil, true, false, true
+				}
+				continue
+			}
+			txs = append(txs, t)
+			continue
+		}
+
+		txs = append(txs, t)
+	}
+
+	return txs, matched, false, false
+}
+
+func prepareTableMatchers(filter BinlogFilter) (includeMatcher, excludeMatcher *tableMatcher, err error) {
+	if len(filter.IncludeTables) > 0 {
+		includeMatcher, err = buildTableMatcher(filter.IncludeTables)
+		if err != nil {
+			return nil, nil, fmt.Errorf("invalid include-tables: %w", err)
+		}
+	}
+	if len(filter.ExcludeTables) > 0 {
+		excludeMatcher, err = buildTableMatcher(filter.ExcludeTables)
+		if err != nil {
+			return nil, nil, fmt.Errorf("invalid exclude-tables: %w", err)
+		}
+	}
+	return includeMatcher, excludeMatcher, nil
+}
+
+func buildTableMatcher(patterns []string) (*tableMatcher, error) {
+	m := &tableMatcher{
+		exactMatch: make(map[string]bool),
+		dbWildcard: make(map[string]bool),
+		tbWildcard: make(map[string]bool),
+	}
+
+	for _, pattern := range patterns {
+		origin := pattern
+		pattern = strings.ToLower(strings.TrimSpace(pattern))
+		if pattern == "" {
+			continue
+		}
+		if pattern == "*.*" {
+			m.matchAll = true
+			continue
+		}
+		parts := strings.Split(pattern, ".")
+		if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+			return nil, fmt.Errorf("invalid table pattern %q: expect db.table", strings.TrimSpace(origin))
+		}
+		db, tb := parts[0], parts[1]
+		if db != "*" && strings.Contains(db, "*") {
+			return nil, fmt.Errorf("invalid table pattern %q: wildcard '*' must occupy full db segment", strings.TrimSpace(origin))
+		}
+		if tb != "*" && strings.Contains(tb, "*") {
+			return nil, fmt.Errorf("invalid table pattern %q: wildcard '*' must occupy full table segment", strings.TrimSpace(origin))
+		}
+		if db == "*" && tb == "*" {
+			m.matchAll = true
+		} else if db == "*" {
+			m.tbWildcard[tb] = true
+		} else if tb == "*" {
+			m.dbWildcard[db] = true
+		} else {
+			m.exactMatch[db+"."+tb] = true
+		}
+	}
+	return m, nil
+}
+
+func hasConfiguredTablePatterns(patterns []string) bool {
+	for _, p := range patterns {
+		if strings.TrimSpace(p) != "" {
+			return true
+		}
+	}
+	return false
+}
+
+func recomputeTxStatsFromVisibleDetails(tx *Transaction) {
+	if tx == nil {
+		return
+	}
+	if len(tx.Txs) == 0 {
+		tx.RowsCount = 0
+		tx.Size = 0
+		return
+	}
+
+	firstSet := false
+	minStart := 0
+	maxEnd := 0
+	rows := 0
+	for _, d := range tx.Txs {
+		rows += d.RowCount
+		if !firstSet {
+			minStart = d.StartPos
+			maxEnd = d.EndPos
+			firstSet = true
+			continue
+		}
+		if d.StartPos < minStart {
+			minStart = d.StartPos
+		}
+		if d.EndPos > maxEnd {
+			maxEnd = d.EndPos
+		}
+	}
+
+	tx.RowsCount = rows
+	tx.StartPos = minStart
+	tx.EndPos = maxEnd
+	if maxEnd > minStart {
+		tx.Size = maxEnd - minStart
+	} else {
+		tx.Size = 0
+	}
+}
+
+func equalFoldShort(s, lower string) bool {
+	if len(s) != len(lower) {
+		return false
+	}
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		if 'A' <= c && c <= 'Z' {
+			c += 'a' - 'A'
+		}
+		if c != lower[i] {
+			return false
+		}
+	}
+	return true
+}
@@ -0,0 +1,85 @@
+package binlog
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestBuildTableMatcher_InvalidPattern(t *testing.T) {
+	cases := [][]string{
+		{"db"},
+		{"db."},
+		{".tb"},
+		{"db.tb.more"},
+		{"db.t*"},
+		{"d*.tb"},
+	}
+
+	for _, patterns := range cases {
+		if _, err := buildTableMatcher(patterns); err == nil {
+			t.Fatalf("expected invalid pattern error, got nil: %v", patterns)
+		}
+	}
+}
+
+func TestPrepareTableMatchers_ReturnErrorOnInvalidPattern(t *testing.T) {
+	_, _, err := prepareTableMatchers(BinlogFilter{IncludeTables: []string{"invalid"}})
+	if err == nil {
+		t.Fatal("expected include-tables error, got nil")
+	}
+	if !strings.Contains(err.Error(), "invalid include-tables") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestSelectVisibleTxDetails_IncludeBlank(t *testing.T) {
+	includeMatcher, err := buildTableMatcher([]string{"db1.tb1"})
+	if err != nil {
+		t.Fatalf("build include matcher failed: %v", err)
+	}
+
+	tx := Transaction{Txs: []TxDetail{
+		{SqlType: "query", Sql: "BEGIN"},
+		{SqlType: "insert", Db: "db1", Table: "tb1", RowCount: 1},
+	}}
+
+	txs, matched, pickAll, skipAll := selectVisibleTxDetails(tx, includeMatcher, nil, BinlogFilter{IncludeBlank: true})
+	if !matched {
+		t.Fatal("expected matched=true")
+	}
+	if pickAll || skipAll {
+		t.Fatalf("unexpected pickAll/skipAll: %v/%v", pickAll, skipAll)
+	}
+	if len(txs) != 2 {
+		t.Fatalf("expected 2 details with IncludeBlank=true, got %d", len(txs))
+	}
+	if txs[0].SqlType != "query" || txs[1].Table != "tb1" {
+		t.Fatalf("unexpected details order/content: %#v", txs)
+	}
+}
+
+func TestSelectVisibleTxDetails_ExcludeBlank(t *testing.T) {
+	excludeMatcher, err := buildTableMatcher([]string{"db2.tb2"})
+	if err != nil {
+		t.Fatalf("build exclude matcher failed: %v", err)
+	}
+
+	tx := Transaction{Txs: []TxDetail{
+		{SqlType: "query", Sql: "BEGIN"},
+		{SqlType: "insert", Db: "db1", Table: "tb1", RowCount: 1},
+	}}
+
+	txs, matched, pickAll, skipAll := selectVisibleTxDetails(tx, nil, excludeMatcher, BinlogFilter{ExcludeBlank: true})
+	if !matched {
+		t.Fatal("expected matched=true when excluding blank detail")
+	}
+	if pickAll || skipAll {
+		t.Fatalf("unexpected pickAll/skipAll: %v/%v", pickAll, skipAll)
+	}
+	if len(txs) != 1 {
+		t.Fatalf("expected 1 detail after ExcludeBlank=true, got %d", len(txs))
+	}
+	if txs[0].Db != "db1" || txs[0].Table != "tb1" {
+		t.Fatalf("unexpected remaining detail: %#v", txs[0])
+	}
+}
@@ -0,0 +1,150 @@
+package binlog
+
+import (
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+
+	"github.com/starainrt/go-mysql/replication"
+)
+
+func validateBinlogHeader(f *os.File) error {
+	const fileTypeBytes = int64(4)
+	b := make([]byte, fileTypeBytes)
+
+	if _, err := f.Read(b); err != nil {
+		return fmt.Errorf("read binlog header failed: %w", err)
+	}
+	if !bytes.Equal(b, replication.BinLogFileHeader) {
+		return ErrInvalidBinlogHeader
+	}
+	if _, err := f.Seek(fileTypeBytes, io.SeekStart); err != nil {
+		return fmt.Errorf("seek after header failed: %w", err)
+	}
+	return nil
+}
+
+func readEventHeader(r io.Reader, parser *replication.BinlogParser, headBuf []byte) (*replication.EventHeader, error) {
+	if _, err := io.ReadFull(r, headBuf); err != nil {
+		return nil, err
+	}
+	h, err := parser.ParseHeader(headBuf)
+	if err != nil {
+		return nil, fmt.Errorf("parse header failed: %w", err)
+	}
+	if h.EventSize <= uint32(replication.EventHeaderSize) {
+		return nil, fmt.Errorf("%w: event size is %d", ErrEventTooSmall, h.EventSize)
+	}
+	return h, nil
+}
+
+func readEventBody(r io.Reader, h *replication.EventHeader) ([]byte, error) {
+	bodyLen := int(h.EventSize) - replication.EventHeaderSize
+	body := make([]byte, bodyLen)
+	if _, err := io.ReadFull(r, body); err != nil {
+		return nil, fmt.Errorf("read event body failed: %w (need %d bytes)", err, bodyLen)
+	}
+	return body, nil
+}
+
+func skipEventBody(r io.Reader, h *replication.EventHeader) error {
+	bodyLen := int64(h.EventSize) - int64(replication.EventHeaderSize)
+	if bodyLen <= 0 {
+		return nil
+	}
+	if _, err := io.CopyN(io.Discard, r, bodyLen); err != nil {
+		return fmt.Errorf("skip event body failed: %w", err)
+	}
+	return nil
+}
+
+var rawDataPool = sync.Pool{
+	New: func() any {
+		b := make([]byte, 0, 64*1024)
+		return &b
+	},
+}
+
+func getRawDataBuf(n int) []byte {
+	p := rawDataPool.Get().(*[]byte)
+	if cap(*p) < n {
+		return make([]byte, n)
+	}
+	return (*p)[:n]
+}
+
+func putRawDataBuf(b []byte) {
+	if cap(b) > maxPooledRawDataCap {
+		return
+	}
+	b = b[:0]
+	rawDataPool.Put(&b)
+}
+
+func formatBodyPreview(body []byte, maxBytes int) string {
+	if maxBytes <= 0 {
+		maxBytes = 256
+	}
+	if len(body) == 0 {
+		return "len=0"
+	}
+	previewLen := len(body)
+	truncated := false
+	if previewLen > maxBytes {
+		previewLen = maxBytes
+		truncated = true
+	}
+	hexBody := hex.EncodeToString(body[:previewLen])
+	if truncated {
+		return fmt.Sprintf("len=%d preview(hex,%dB)=%s...", len(body), previewLen, hexBody)
+	}
+	return fmt.Sprintf("len=%d preview(hex,%dB)=%s", len(body), previewLen, hexBody)
+}
+
+func parseEvent(parser *replication.BinlogParser, h *replication.EventHeader, headBuf []byte, body []byte) (replication.Event, error) {
+	rawLen := len(headBuf) + len(body)
+	rawData := getRawDataBuf(rawLen)
+	copy(rawData, headBuf)
+	copy(rawData[len(headBuf):], body)
+
+	e, err := parser.ParseEvent(h, body, rawData)
+	putRawDataBuf(rawData)
+	if err != nil {
+		return nil, fmt.Errorf("parse event failed at pos %d: Header %+v, Body %s, Err: %w",
+			h.LogPos, h, formatBodyPreview(body, 256), err)
+	}
+	return e, nil
+}
+
+func seekToPosition(f *os.File, parser *replication.BinlogParser, pos int64) error {
+	if err := validateBinlogHeader(f); err != nil {
+		return err
+	}
+
+	headBuf := make([]byte, replication.EventHeaderSize)
+	for {
+		h, err := readEventHeader(f, parser, headBuf)
+		if err != nil {
+			return fmt.Errorf("seek to position failed: %w", err)
+		}
+		body, err := readEventBody(f, h)
+		if err != nil {
+			return err
+		}
+		_, err = parseEvent(parser, h, headBuf, body)
+		if err != nil {
+			return err
+		}
+		if h.EventType == replication.FORMAT_DESCRIPTION_EVENT || h.EventType == replication.GTID_EVENT {
+			break
+		}
+	}
+
+	if _, err := f.Seek(pos, io.SeekStart); err != nil {
+		return fmt.Errorf("seek to pos %d failed: %w", pos, err)
+	}
+	return nil
+}
@@ -0,0 +1,33 @@
+package binlog
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestFormatBodyPreview(t *testing.T) {
+	if got := formatBodyPreview(nil, 256); got != "len=0" {
+		t.Fatalf("unexpected empty preview: %q", got)
+	}
+
+	small := []byte{0x01, 0x02, 0xAB}
+	got := formatBodyPreview(small, 8)
+	if !strings.Contains(got, "len=3") || !strings.Contains(got, "0102ab") {
+		t.Fatalf("unexpected preview for small body: %q", got)
+	}
+	if strings.Contains(got, "...") {
+		t.Fatalf("small body should not be truncated: %q", got)
+	}
+
+	large := make([]byte, 300)
+	for i := range large {
+		large[i] = byte(i)
+	}
+	got = formatBodyPreview(large, 16)
+	if !strings.Contains(got, "len=300") || !strings.Contains(got, "preview(hex,16B)=") {
+		t.Fatalf("unexpected preview for large body: %q", got)
+	}
+	if !strings.HasSuffix(got, "...") {
+		t.Fatalf("large body should be truncated with ellipsis: %q", got)
+	}
+}
@@ -0,0 +1,57 @@
+package binlog
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestParseBinlogWithOptions_ProgressAndStop(t *testing.T) {
+	t.Skip("skips large-binlog integration test; this standalone module must not depend on external sample files")
+}
+
+func TestParseBinlogWithOptions_ContextCancel(t *testing.T) {
+	path := "./test/mysql-bin56.000003"
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := ParseBinlogWithOptions(path, ParseOptions{Context: ctx}, func(Transaction) bool {
+		t.Fatal("transaction callback should not run after context cancellation")
+		return false
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+}
+
+func TestParseBinlogWithOptions_ParseBinSample(t *testing.T) {
+	t.Skip("skips large-binlog integration test; this standalone module must not depend on external sample files")
+}
+
+func TestParseBinlogWithOptions_LargeBinProgressStop(t *testing.T) {
+	t.Skip("skips large-binlog integration test; this standalone module must not depend on external sample files")
+}
+
+func TestParseBinlogWithOptions_ExcludeAllGTIDDoesNotEmitEmptyTransaction(t *testing.T) {
+	t.Skip("skips large-binlog integration test; this standalone module must not depend on external sample files")
+}
+
+func TestTransactionJSONPreservesZeroLogicalClockValues(t *testing.T) {
+	raw, err := json.Marshal(Transaction{
+		GTID:           "uuid:1",
+		LastCommitted:  0,
+		SequenceNumber: 1,
+	})
+	if err != nil {
+		t.Fatalf("marshal transaction failed: %v", err)
+	}
+	doc := string(raw)
+	if !strings.Contains(doc, `"lastCommitted":0`) {
+		t.Fatalf("expected zero lastCommitted to be present, got %s", doc)
+	}
+	if !strings.Contains(doc, `"sequenceNumber":1`) {
+		t.Fatalf("expected sequenceNumber to be present, got %s", doc)
+	}
+}
@@ -0,0 +1,38 @@
+package binlog
+
+import (
+	"time"
+)
+
+func ParseBinlogFile(path string, fx func(transaction Transaction) bool) error {
+	return ParseBinlogWithOptions(path, ParseOptions{}, fx)
+}
+
+func finalizeTx(tx *Transaction, onlyShowGtid bool) {
+	idx := 0
+	for k, v := range tx.Txs {
+		if v.SqlType != "query" && len(tx.sqlOrigin) > idx {
+			v.Sql = tx.sqlOrigin[idx]
+			idx++
+		}
+		tx.RowsCount += v.RowCount
+		tx.Txs[k] = v
+	}
+
+	if onlyShowGtid {
+		tx.Size = 0
+	} else {
+		tx.Size = tx.EndPos - tx.StartPos
+	}
+}
+
+func fillTimeLazy(tx *Transaction) {
+	if tx.Timestamp != 0 && tx.Time.IsZero() {
+		tx.Time = time.Unix(tx.Timestamp, 0)
+	}
+	for i := range tx.Txs {
+		if tx.Txs[i].Timestamp != 0 && tx.Txs[i].Time.IsZero() {
+			tx.Txs[i].Time = time.Unix(tx.Txs[i].Timestamp, 0)
+		}
+	}
+}
@@ -0,0 +1,157 @@
+package binlog
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"time"
+)
+
+var (
+	ErrInvalidBinlogHeader = errors.New("invalid binlog file header")
+	ErrEventTooSmall       = errors.New("event size too small")
+)
+
+const (
+	CompressionNone uint64 = 255
+	CompressionZSTD uint64 = 0
+)
+
+const (
+	maxPooledRawDataCap = 4 << 20 // 4MB
+	defaultReadBufSize  = 1 << 20 // 1MB
+)
+
+type TxDetail struct {
+	StartPos           int             `json:"startPos"`
+	EndPos             int             `json:"endPos"`
+	RowCount           int             `json:"rowCount"`
+	Timestamp          int64           `json:"timestamp"`
+	Time               time.Time       `json:"time"`
+	Sql                string          `json:"sql"`
+	Db                 string          `json:"db"`
+	Table              string          `json:"table"`
+	SqlType            string          `json:"sqlType"`
+	CompressionType    string          `json:"compressionType"`
+	Rows               [][]interface{} `json:"rows"`
+	ColumnTypes        []int           `json:"columnTypes,omitempty"`
+	ColumnCollationIDs []uint64        `json:"columnCollationIds,omitempty"`
+}
+
+const (
+	STATUS_PREPARE uint8 = iota
+	STATUS_BEGIN
+	STATUS_COMMIT
+	STATUS_ROLLBACK
+)
+
+type Transaction struct {
+	GTID                     string     `json:"gtid"`
+	Timestamp                int64      `json:"timestamp"`
+	Time                     time.Time  `json:"time"`
+	StartPos                 int        `json:"startPos"`
+	EndPos                   int        `json:"endPos"`
+	Size                     int        `json:"size"`
+	RowsCount                int        `json:"rowsCount"`
+	Status                   uint8      `json:"status"`
+	TxStartTime              int64      `json:"txStartTime"`
+	TxEndTime                int64      `json:"txEndTime"`
+	LastCommitted            int64      `json:"lastCommitted"`
+	SequenceNumber           int64      `json:"sequenceNumber"`
+	TransactionLength        uint64     `json:"transactionLength,omitempty"`
+	ImmediateCommitTimestamp uint64     `json:"immediateCommitTimestamp,omitempty"`
+	OriginalCommitTimestamp  uint64     `json:"originalCommitTimestamp,omitempty"`
+	sqlOrigin                []string   `json:"sqlOrigin"`
+	Txs                      []TxDetail `json:"txs"`
+	dmlEventCount            int
+}
+
+func (t Transaction) GetSqlOrigin() []string {
+	return t.sqlOrigin
+}
+
+type BinlogFilter struct {
+	IncludeGtid      string
+	ExcludeGtid      string
+	IncludeTables    []string
+	ExcludeTables    []string
+	StartPos         int
+	EndPos           int
+	StartDate        time.Time
+	EndDate          time.Time
+	BigThan          int
+	SmallThan        int
+	OnlyShowGtid     bool
+	OnlyShowDML      bool
+	PickTxAllIfMatch bool
+	ExcludeBlank     bool
+	IncludeBlank     bool
+}
+
+type BinlogEvent struct {
+	Type                     string
+	DB                       string
+	TB                       string
+	Data                     string
+	RowCnt                   uint32
+	Rows                     [][]interface{}
+	ColumnTypes              []int
+	ColumnCollationIDs       []uint64
+	CompressionType          string
+	EventType                byte
+	ServerID                 uint32
+	Timestamp                uint32
+	LogPos                   uint32
+	EventSize                uint32
+	LastCommitted            int64
+	SequenceNumber           int64
+	TransactionLength        uint64
+	ImmediateCommitTimestamp uint64
+	OriginalCommitTimestamp  uint64
+}
+
+type ParseProgress struct {
+	Path     string
+	Event    BinlogEvent
+	EventPos int64
+	NextPos  int64
+	FileSize int64
+}
+
+type ParseOptions struct {
+	Context    context.Context
+	Filter     BinlogFilter
+	StartPos   int64
+	OnProgress func(ParseProgress) bool
+}
+
+type tableMatcher struct {
+	exactMatch map[string]bool
+	dbWildcard map[string]bool
+	tbWildcard map[string]bool
+	matchAll   bool
+}
+
+func (m *tableMatcher) match(db, tb string) bool {
+	db = strings.ToLower(strings.TrimSpace(db))
+	tb = strings.ToLower(strings.TrimSpace(tb))
+
+	if m.matchAll {
+		return true
+	}
+	if m.dbWildcard[db] || m.tbWildcard[tb] {
+		return true
+	}
+	if len(m.exactMatch) > 0 {
+		// Go 1.12+ 对 map[string] 查找时 string([]byte) 不分配
+		var buf [128]byte
+		key := buf[:0]
+		key = append(key, db...)
+		key = append(key, '.')
+		key = append(key, tb...)
+		if m.exactMatch[string(key)] {
+			return true
+		}
+	}
+	return false
+}
@@ -0,0 +1,239 @@
+package binlog
+
+import (
+	"strings"
+	"time"
+)
+
+type TransactionOutcome string
+
+const (
+	TransactionOutcomeCommit     TransactionOutcome = "commit"
+	TransactionOutcomeRollback   TransactionOutcome = "rollback"
+	TransactionOutcomeAutocommit TransactionOutcome = "autocommit"
+	TransactionOutcomeOpen       TransactionOutcome = "open"
+	TransactionOutcomeOther      TransactionOutcome = "other"
+)
+
+type TransactionSummary struct {
+	GTID                     string
+	SeenTime                 time.Time
+	LastEventTime            time.Time
+	BeginTime                time.Time
+	EndTime                  time.Time
+	Duration                 time.Duration
+	HasBeginBoundary         bool
+	HasEndBoundary           bool
+	HasExplicitDuration      bool
+	HasDuration              bool
+	Outcome                  TransactionOutcome
+	StartPos                 int
+	EndPos                   int
+	Size                     int
+	RowsCount                int
+	StatementsCount          int
+	Tables                   []string
+	SampleSQL                string
+	LastCommitted            int64
+	SequenceNumber           int64
+	TransactionLength        uint64
+	ImmediateCommitTimestamp uint64
+	OriginalCommitTimestamp  uint64
+}
+
+func SummarizeTransaction(tx Transaction) TransactionSummary {
+	fillTimeLazy(&tx)
+	s := TransactionSummary{
+		GTID:                     strings.TrimSpace(tx.GTID),
+		SeenTime:                 tx.Time,
+		LastEventTime:            tx.Time,
+		StartPos:                 tx.StartPos,
+		EndPos:                   tx.EndPos,
+		Size:                     tx.Size,
+		RowsCount:                tx.RowsCount,
+		LastCommitted:            tx.LastCommitted,
+		SequenceNumber:           tx.SequenceNumber,
+		TransactionLength:        tx.TransactionLength,
+		ImmediateCommitTimestamp: tx.ImmediateCommitTimestamp,
+		OriginalCommitTimestamp:  tx.OriginalCommitTimestamp,
+	}
+
+	tableSeen := make(map[string]struct{}, 8)
+	hasBegin := false
+	hasCommit := false
+	hasRollback := false
+	hasNonBoundary := false
+
+	for _, detail := range tx.Txs {
+		if detail.Time.IsZero() && detail.Timestamp != 0 {
+			detail.Time = time.Unix(detail.Timestamp, 0)
+		}
+		if !detail.Time.IsZero() {
+			s.LastEventTime = maxSummaryTime(s.LastEventTime, detail.Time)
+			switch boundaryKind(detail.Sql) {
+			case "begin":
+				s.BeginTime = minSummaryTime(s.BeginTime, detail.Time)
+				s.HasBeginBoundary = true
+			case "commit", "rollback":
+				s.EndTime = maxSummaryTime(s.EndTime, detail.Time)
+				s.HasEndBoundary = true
+			}
+		}
+		s.RowsCount += detail.RowCount
+		if strings.TrimSpace(detail.Sql) != "" && !isBoundaryDetail(detail) {
+			s.StatementsCount++
+			if s.SampleSQL == "" {
+				s.SampleSQL = compactSampleSQL(detail.Sql)
+			}
+		}
+		tableKey := summaryTableKey(detail.Db, detail.Table)
+		if tableKey != "" {
+			if _, ok := tableSeen[tableKey]; !ok {
+				tableSeen[tableKey] = struct{}{}
+				s.Tables = append(s.Tables, tableKey)
+			}
+		}
+		switch boundaryKind(detail.Sql) {
+		case "begin":
+			hasBegin = true
+		case "commit":
+			hasCommit = true
+		case "rollback":
+			hasRollback = true
+		default:
+			if !strings.EqualFold(strings.TrimSpace(detail.SqlType), "query") || strings.TrimSpace(detail.Sql) != "" {
+				hasNonBoundary = true
+			}
+		}
+	}
+	if tx.RowsCount > 0 {
+		s.RowsCount = tx.RowsCount
+	}
+	if s.SeenTime.IsZero() {
+		s.SeenTime = firstDetailTime(tx)
+	}
+	if s.BeginTime.IsZero() {
+		s.BeginTime = firstNonZeroSummaryTime(firstDetailTime(tx), s.SeenTime)
+	}
+	if s.EndTime.IsZero() {
+		s.EndTime = firstNonZeroSummaryTime(s.LastEventTime, s.BeginTime, s.SeenTime)
+	}
+	if !s.BeginTime.IsZero() && !s.EndTime.IsZero() && !s.EndTime.Before(s.BeginTime) {
+		s.Duration = s.EndTime.Sub(s.BeginTime)
+		s.HasDuration = true
+		s.HasExplicitDuration = s.HasBeginBoundary && s.HasEndBoundary
+	}
+	s.Outcome = summarizeOutcome(tx, hasBegin, hasCommit, hasRollback, hasNonBoundary)
+	return s
+}
+
+func summarizeOutcome(tx Transaction, hasBegin bool, hasCommit bool, hasRollback bool, hasNonBoundary bool) TransactionOutcome {
+	switch {
+	case hasCommit:
+		return TransactionOutcomeCommit
+	case hasRollback:
+		return TransactionOutcomeRollback
+	case hasBegin:
+		return TransactionOutcomeOpen
+	}
+	switch tx.Status {
+	case STATUS_COMMIT:
+		return TransactionOutcomeCommit
+	case STATUS_ROLLBACK:
+		return TransactionOutcomeRollback
+	case STATUS_BEGIN:
+		return TransactionOutcomeOpen
+	case STATUS_PREPARE:
+		if hasNonBoundary {
+			return TransactionOutcomeAutocommit
+		}
+	}
+	if hasNonBoundary {
+		return TransactionOutcomeAutocommit
+	}
+	return TransactionOutcomeOther
+}
+
+func isBoundaryDetail(detail TxDetail) bool {
+	if !strings.EqualFold(strings.TrimSpace(detail.SqlType), "query") {
+		return false
+	}
+	return boundaryKind(detail.Sql) != ""
+}
+
+func boundaryKind(sql string) string {
+	switch strings.ToLower(strings.TrimSpace(sql)) {
+	case "begin":
+		return "begin"
+	case "commit":
+		return "commit"
+	case "rollback":
+		return "rollback"
+	default:
+		return ""
+	}
+}
+
+func summaryTableKey(db string, table string) string {
+	db = strings.ToLower(strings.TrimSpace(db))
+	table = strings.ToLower(strings.TrimSpace(table))
+	if db == "" || table == "" {
+		return ""
+	}
+	return db + "." + table
+}
+
+func compactSampleSQL(sql string) string {
+	sql = strings.Join(strings.Fields(sql), " ")
+	if len(sql) <= 200 {
+		return sql
+	}
+	return sql[:197] + "..."
+}
+
+func firstDetailTime(tx Transaction) time.Time {
+	var ret time.Time
+	for _, detail := range tx.Txs {
+		t := detail.Time
+		if t.IsZero() && detail.Timestamp != 0 {
+			t = time.Unix(detail.Timestamp, 0)
+		}
+		ret = minSummaryTime(ret, t)
+	}
+	return ret
+}
+
+func minSummaryTime(a time.Time, b time.Time) time.Time {
+	if a.IsZero() {
+		return b
+	}
+	if b.IsZero() {
+		return a
+	}
+	if b.Before(a) {
+		return b
+	}
+	return a
+}
+
+func maxSummaryTime(a time.Time, b time.Time) time.Time {
+	if a.IsZero() {
+		return b
+	}
+	if b.IsZero() {
+		return a
+	}
+	if b.After(a) {
+		return b
+	}
+	return a
+}
+
+func firstNonZeroSummaryTime(items ...time.Time) time.Time {
+	for _, item := range items {
+		if !item.IsZero() {
+			return item
+		}
+	}
+	return time.Time{}
+}
@@ -0,0 +1,66 @@
+package binlog
+
+import (
+	"testing"
+	"time"
+)
+
+func TestSummarizeTransaction(t *testing.T) {
+	tx := Transaction{
+		GTID:              "uuid:1",
+		Timestamp:         100,
+		StartPos:          120,
+		EndPos:            240,
+		Size:              120,
+		LastCommitted:     98,
+		SequenceNumber:    100,
+		TransactionLength: 4096,
+		Txs: []TxDetail{
+			{SqlType: "query", Sql: "BEGIN", Timestamp: 100},
+			{SqlType: "insert", Db: "Shop", Table: "Orders", Sql: "insert into orders values (1)", RowCount: 2, Timestamp: 101},
+			{SqlType: "delete", Db: "shop", Table: "orders", Sql: "delete from orders where id = 2", RowCount: 1, Timestamp: 102},
+			{SqlType: "query", Sql: "COMMIT", Timestamp: 103},
+		},
+	}
+
+	got := SummarizeTransaction(tx)
+	if got.Outcome != TransactionOutcomeCommit {
+		t.Fatalf("unexpected outcome: %s", got.Outcome)
+	}
+	if !got.HasBeginBoundary || !got.HasEndBoundary || !got.HasExplicitDuration || !got.HasDuration {
+		t.Fatalf("expected explicit duration markers: %#v", got)
+	}
+	if got.Duration != 3*time.Second {
+		t.Fatalf("unexpected duration: %s", got.Duration)
+	}
+	if got.RowsCount != 3 || got.StatementsCount != 2 {
+		t.Fatalf("unexpected rows/statements: rows=%d stmts=%d", got.RowsCount, got.StatementsCount)
+	}
+	if len(got.Tables) != 1 || got.Tables[0] != "shop.orders" {
+		t.Fatalf("unexpected tables: %#v", got.Tables)
+	}
+	if got.SampleSQL != "insert into orders values (1)" {
+		t.Fatalf("unexpected sample sql: %q", got.SampleSQL)
+	}
+	if got.LastCommitted != 98 || got.SequenceNumber != 100 || got.TransactionLength != 4096 {
+		t.Fatalf("logical metadata not preserved: %#v", got)
+	}
+}
+
+func TestSummarizeTransaction_Autocommit(t *testing.T) {
+	tx := Transaction{
+		GTID:      "uuid:2",
+		Timestamp: 100,
+		Txs: []TxDetail{
+			{SqlType: "insert", Db: "db", Table: "tb", Sql: "insert into tb values (1)", RowCount: 1, Timestamp: 100},
+		},
+	}
+
+	got := SummarizeTransaction(tx)
+	if got.Outcome != TransactionOutcomeAutocommit {
+		t.Fatalf("unexpected outcome: %s", got.Outcome)
+	}
+	if got.SeenTime.IsZero() || got.BeginTime.IsZero() || got.EndTime.IsZero() {
+		t.Fatalf("expected lazy times to be filled: %#v", got)
+	}
+}