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

- 暴露事务级 last_committed、sequence_number、transaction_length 和 commit timestamp
- 在 GTID event 转换时透传 logical clock 元数据
- 新增 ParseOptions、ParseProgress，支持上下文取消和解析进度回调
- 保留 TransactionPayloadEvent 展开后的 tablemap 与压缩类型信息
- 增加 TransactionSummary 辅助结构，便于上层统计事务结果、耗时、表分布和逻辑时钟
- 清理测试对外部大 binlog 样本的依赖，保证独立库测试可运行
- 修正跨平台测试与 GTID 输出格式断言，提升发版稳定性

dump_binlog_from_pos_test.go

@@ -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)

get_all_gtid_of_binlog_dir_test.go

@@ -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)
 	}
 }

get_gtid_of_binlog_test.go

@@ -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, ",")
+}

go.mod

@@ -1,26 +1,31 @@
 module b612.me/mysql/binlog
 
-go 1.20
+go 1.24.0
 
 require (
 	b612.me/mysql/gtid v0.0.0-20230425105031-298e51a68044
-	b612.me/staros v1.1.6
-	github.com/starainrt/go-mysql v0.0.0-20230425104003-c669a16617e3
+	b612.me/staros v1.1.8
+	github.com/starainrt/go-mysql v0.0.2
 )
 
 require (
-	b612.me/stario v0.0.8 // indirect
-	b612.me/win32api v0.0.1 // indirect
-	b612.me/wincmd v0.0.2 // indirect
-	github.com/DataDog/zstd v1.5.2 // indirect
-	github.com/Masterminds/semver v1.5.0 // indirect
-	github.com/google/uuid v1.3.0 // indirect
-	github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 // indirect
-	github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 // indirect
-	github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726 // indirect
-	github.com/siddontang/go-log v0.0.0-20190221022429-1e957dd83bed // indirect
-	go.uber.org/atomic v1.7.0 // indirect
-	golang.org/x/crypto v0.0.0-20220321153916-2c7772ba3064 // indirect
-	golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 // indirect
-	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
+	b612.me/stario v0.0.10 // indirect
+	b612.me/win32api v0.0.3 // indirect
+	b612.me/wincmd v0.0.4 // indirect
+	filippo.io/edwards25519 v1.2.0 // indirect
+	github.com/goccy/go-json v0.10.5 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/klauspost/compress v1.18.4 // indirect
+	github.com/pingcap/errors v0.11.5-0.20250523034308-74f78ae071ee // indirect
+	github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a // indirect
+	github.com/pingcap/tidb/pkg/parser v0.0.0-20260219190905-9b9281fa8d6d // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.1 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
+	golang.org/x/term v0.40.0 // indirect
+	golang.org/x/text v0.34.0 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 )

go.sum

@@ -1,133 +1,150 @@
 b612.me/mysql/gtid v0.0.0-20230425105031-298e51a68044 h1:sJrYUl9Sb1tij6ROahFE3r/36Oag3kI92OXDjOKsdwA=
 b612.me/mysql/gtid v0.0.0-20230425105031-298e51a68044/go.mod h1:3EHq1jvlm3a92UxagMjfqSSVYb3KW2H3aT5nd4SiD94=
-b612.me/stario v0.0.7/go.mod h1:or4ssWcxQSjMeu+hRKEgtp0X517b3zdlEOAms8Qscvw=
-b612.me/stario v0.0.8 h1:kaA4pszAKLZJm2D9JmiuYSpgjTeE3VaO74vm+H0vBGM=
-b612.me/stario v0.0.8/go.mod h1:or4ssWcxQSjMeu+hRKEgtp0X517b3zdlEOAms8Qscvw=
-b612.me/staros v1.1.6 h1:m3QaEmPyvPcJVomjWs8cDeauDYFNKv7cLHTiOHClKqM=
-b612.me/staros v1.1.6/go.mod h1:O657LC3qag4VSsHNmt5RM8gKJvzoEGq8IF8WegcRgq0=
-b612.me/win32api v0.0.1 h1:vLFB1xhO6pd9+zB2EyaapKB459Urv3v+C1YwgwOFEWo=
-b612.me/win32api v0.0.1/go.mod h1:MHu0JBQjzxQ2yxpZPUBbn5un45o67eF5iWKa4Q9e0yE=
-b612.me/wincmd v0.0.2 h1:Ub1WtelVT6a3vD4B6zDYo3UPO/t9ymnI3x1dQPJcrGw=
-b612.me/wincmd v0.0.2/go.mod h1:bwpyCKfSDY8scSMo3Lrd0Qnqvpz7/CILL7oodfG0wgo=
-github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/DataDog/zstd v1.5.2 h1:vUG4lAyuPCXO0TLbXvPv7EB7cNK1QV/luu55UHLrrn8=
-github.com/DataDog/zstd v1.5.2/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
-github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
-github.com/Masterminds/semver v1.5.0/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
+b612.me/stario v0.0.10 h1:+cIyiDCBCjUfodMJDp4FLs+2E1jo7YENkN+sMEe6550=
+b612.me/stario v0.0.10/go.mod h1:1Owmu9jzKWgs4VsmeI8YWlGwLrCwPNM/bYpxkyn+MMk=
+b612.me/staros v1.1.8 h1:5Bpuf9q2nH75S2ekmieJuH3Y8LTqg/voxXCOiMAC3kk=
+b612.me/staros v1.1.8/go.mod h1:4KmokjKXFW5h1hbA4aIv5O+2FptVzBubCo7IPirfqm8=
+b612.me/win32api v0.0.2/go.mod h1:sj66sFJDKElEjOR+0YhdSW6b4kq4jsXu4T5/Hnpyot0=
+b612.me/win32api v0.0.3 h1:TfINlv9BBmWC/YbkJ0MTpN1NzTFPnnTGB5Dux6iRWIA=
+b612.me/win32api v0.0.3/go.mod h1:sj66sFJDKElEjOR+0YhdSW6b4kq4jsXu4T5/Hnpyot0=
+b612.me/wincmd v0.0.4 h1:fv9p1V8mw2HdUjaoZBWZy0T41JftueyLxAuch1MgtdI=
+b612.me/wincmd v0.0.4/go.mod h1:o3yPoE+DpVPHGKl/q1WT1C8OaIVwHEnpeNgMFqzlwD8=
+filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
+filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
-github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548/go.mod h1:e6NPNENfs9mPDVNRekM7lKScauxd5kXTr1Mfyig6TDM=
-github.com/cznic/sortutil v0.0.0-20181122101858-f5f958428db8/go.mod h1:q2w6Bg5jeox1B+QkJ6Wp/+Vn0G/bo3f1uY7Fn3vivIQ=
-github.com/cznic/strutil v0.0.0-20171016134553-529a34b1c186/go.mod h1:AHHPPPXTw0h6pVabbcbyGRK1DckRn7r/STdZEeIDzZc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
-github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
-github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
-github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/jmoiron/sqlx v1.3.3/go.mod h1:2BljVx/86SuTyjE+aPYlHCTNvZrnJXghYGpNiXLBMCQ=
+github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
+github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c=
+github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
-github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
-github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8 h1:USx2/E1bX46VG32FIw034Au6seQ2fY9NEILmNh/UlQg=
-github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8/go.mod h1:B1+S9LNcuMyLH/4HMTViQOJevkGiik3wW2AN9zb2fNQ=
 github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
-github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 h1:+FZIDR/D97YOPik4N4lPDaUcLDF/EQPogxtlHB2ZZRM=
-github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
-github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
-github.com/pingcap/tidb/parser v0.0.0-20221126021158-6b02a5d8ba7d/go.mod h1:ElJiub4lRy6UZDb+0JHDkGEdr6aOli+ykhyej7VCLoI=
+github.com/pingcap/errors v0.11.5-0.20250523034308-74f78ae071ee h1:/IDPbpzkzA97t1/Z1+C3KlxbevjMeaI6BQYxvivu4u8=
+github.com/pingcap/errors v0.11.5-0.20250523034308-74f78ae071ee/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
+github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a h1:WIhmJBlNGmnCWH6TLMdZfNEDaiU8cFpZe3iaqDbQ0M8=
+github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA=
+github.com/pingcap/tidb/pkg/parser v0.0.0-20260219190905-9b9281fa8d6d h1:jD97s7AVHGuKGqvbJkTcNpMlcSx5Qv/sZF0XHENK+0w=
+github.com/pingcap/tidb/pkg/parser v0.0.0-20260219190905-9b9281fa8d6d/go.mod h1:oHE+ub2QaDERd+UNHe4z2BhFV2jZrm7VNOe6atR9AF4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
-github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24 h1:pntxY8Ary0t43dCZ5dqY4YTJCObLY1kIXl0uzMv+7DE=
-github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
-github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726 h1:xT+JlYxNGqyT+XcU8iUrN18JYed2TvG9yN5ULG2jATM=
-github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726/go.mod h1:3yhqj7WBBfRhbBlzyOC3gUxftwsU0u8gqevxwIHQpMw=
-github.com/siddontang/go-log v0.0.0-20190221022429-1e957dd83bed h1:KMgQoLJGCq1IoZpLZE3AIffh9veYWoVlsvA4ib55TMM=
-github.com/siddontang/go-log v0.0.0-20190221022429-1e957dd83bed/go.mod h1:yFdBgwXP24JziuRl2NMUahT7nGLNOKi1SIiFxMttVD4=
-github.com/starainrt/go-mysql v0.0.0-20230425104003-c669a16617e3 h1:uG2HtVhi/UtC+sUiArLV8cc8SDlUrRkqLChriV5WYZw=
-github.com/starainrt/go-mysql v0.0.0-20230425104003-c669a16617e3/go.mod h1:nNTDdEbTDJ9dK0b1NJ+xo804OQuZw39GfNfpryd9d3w=
+github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
+github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
+github.com/starainrt/go-mysql v0.0.2 h1:PcQX1kNekew1LEqWExN71NEjAWdZ8Cpuau0F1C29JQY=
+github.com/starainrt/go-mysql v0.0.2/go.mod h1:O1TXVhZd1iT/vU027lg5mxAPU5b8UrG1SRkQqa3bgww=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
-github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
-go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
-go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
-go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
-go.uber.org/zap v1.18.1/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
+go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
+go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc=
+go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220321153916-2c7772ba3064 h1:S25/rfnfsMVgORT4/J61MJ7rdyseOZOyvLIrZEZ7s6s=
-golang.org/x/crypto v0.0.0-20220321153916-2c7772ba3064/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/exp v0.0.0-20181106170214-d68db9428509/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
+golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
+golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
+golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
+golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts=
+golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
-golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
+golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
+golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 h1:OH54vjqzRWmbJ62fjuhxy7AxFFgoHN0/DPc/UrL8cAs=
-golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
+golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
+golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
+golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
+golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
+golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg=
+golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
+golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20201125231158-b5590deeca9b/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
+golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
+golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
+gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
+gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-modernc.org/fileutil v1.0.0/go.mod h1:JHsWpkrk/CnVV1H/eGlFf85BEpfkrp56ro8nojIq9Q8=
-modernc.org/golex v1.0.1/go.mod h1:QCA53QtsT1NdGkaZZkF5ezFwk4IXh4BGNafAARTC254=
-modernc.org/lex v1.0.0/go.mod h1:G6rxMTy3cH2iA0iXL/HRRv4Znu8MK4higxph/lE7ypk=
-modernc.org/lexer v1.0.0/go.mod h1:F/Dld0YKYdZCLQ7bD0USbWL4YKCyTDRDHiDTOs0q0vk=
-modernc.org/mathutil v1.0.0/go.mod h1:wU0vUrJsVWBZ4P6e7xtFJEhFSNsfRLJ8H458uRjg03k=
-modernc.org/mathutil v1.4.1/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E=
-modernc.org/parser v1.0.0/go.mod h1:H20AntYJ2cHHL6MHthJ8LZzXCdDCHMWt1KZXtIMjejA=
-modernc.org/parser v1.0.2/go.mod h1:TXNq3HABP3HMaqLK7brD1fLA/LfN0KS6JxZn71QdDqs=
-modernc.org/scanner v1.0.1/go.mod h1:OIzD2ZtjYk6yTuyqZr57FmifbM9fIH74SumloSsajuE=
-modernc.org/sortutil v1.0.0/go.mod h1:1QO0q8IlIlmjBIwm6t/7sof874+xCfZouyqZMLIAtxM=
-modernc.org/strutil v1.0.0/go.mod h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs=
-modernc.org/strutil v1.1.0/go.mod h1:lstksw84oURvj9y3tn8lGvRxyRC1S2+g5uuIzNfIOBs=
-modernc.org/y v1.0.1/go.mod h1:Ho86I+LVHEI+LYXoUKlmOMAM1JTXOCfj8qi1T8PsClE=

parse.go

@@ -1,656 +1,8 @@
 package binlog
 
-import (
-	"b612.me/mysql/gtid"
-	"b612.me/staros"
-	"bytes"
-	"fmt"
-	"github.com/starainrt/go-mysql/replication"
-	"io"
-	"os"
-	"strings"
-	"time"
-)
-
-type TxDetail struct {
-	StartPos        int
-	EndPos          int
-	RowCount        int
-	Timestamp       int64
-	Time            time.Time
-	Sql             string
-	Db              string
-	Table           string
-	SqlType         string
-	CompressionType string
-	Rows            [][]interface{}
-}
-
-type Transaction struct {
-	GTID      string
-	Timestamp int64
-	Time      time.Time
-	StartPos  int
-	EndPos    int
-	Size      int
-	RowsCount int
-	sqlOrigin []string
-	Txs       []TxDetail
-}
-
-func (t Transaction) GetSqlOrigin() []string {
-	return t.sqlOrigin
-}
-
-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 f != nil {
-		defer f.Close()
-	}
-	if err != nil {
-		return err
-	}
-
-	fileTypeBytes := int64(4)
-
-	b := make([]byte, fileTypeBytes)
-	// 读取binlog头
-	if _, err = f.Read(b); err != nil {
-		return err
-	} else if !bytes.Equal(b, replication.BinLogFileHeader) {
-		//不是binlog格式
-		return err
-	}
-	// must not seek to other position, otherwise the program may panic because formatevent, table map event is skipped
-	if _, err = f.Seek(fileTypeBytes, os.SEEK_SET); err != nil {
-		return err
-	}
-	return parseBinlogDetail(f, fx)
-}
-
-func parseBinlogDetail(r io.Reader, f func(Transaction) bool) error {
-	parse := replication.NewBinlogParser()
-	parse.SetParseTime(false)
-	parse.SetUseDecimal(false)
-	// process: 0, continue: 1, break: 2, EOF: 3
-	var (
-		err      error
-		n        int64
-		tbMapPos uint32 = 0
-	)
-	var tx Transaction
-	currentGtid := ""
-	for {
-		headBuf := make([]byte, replication.EventHeaderSize)
-		if _, err = io.ReadFull(r, headBuf); err == io.EOF {
-			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
-			}
-			tx.Size = tx.EndPos - tx.StartPos
-			if f != nil {
-				f(tx)
-			}
-			return nil
-		} else if err != nil {
-			return err
-		}
-
-		var h *replication.EventHeader
-		h, err = parse.ParseHeader(headBuf)
-		if err != nil {
-			return err
-		}
-		//fmt.Printf("parsing %s %d %s\n", *binlog, h.LogPos, GetDatetimeStr(int64(h.Timestamp), int64(0), DATETIME_FORMAT))
-
-		if h.EventSize <= uint32(replication.EventHeaderSize) {
-			err = fmt.Errorf("invalid event header, event size is %d, too small", h.EventSize)
-			return err
-		}
-
-		var buf bytes.Buffer
-		if n, err = io.CopyN(&buf, r, int64(h.EventSize)-int64(replication.EventHeaderSize)); err != nil {
-			err = fmt.Errorf("get event body err %v, need %d - %d, but got %d", err, h.EventSize, replication.EventHeaderSize, n)
-			return err
-		}
-
-		//h.Dump(os.Stdout)
-
-		data := buf.Bytes()
-		var rawData []byte
-		rawData = append(rawData, headBuf...)
-		rawData = append(rawData, data...)
-
-		eventLen := int(h.EventSize) - replication.EventHeaderSize
-
-		if len(data) != eventLen {
-			err = fmt.Errorf("invalid data size %d in event %s, less event length %d", len(data), h.EventType, eventLen)
-			return err
-		}
-
-		var e replication.Event
-		e, err = parse.ParseEvent(h, data, rawData)
-		if err != nil {
-			return err
-		}
-		if h.EventType == replication.TABLE_MAP_EVENT {
-			tbMapPos = h.LogPos - h.EventSize // avoid mysqlbing mask the row event as unknown table row event
-		}
-
-		//e.Dump(os.Stdout)
-
-		//binEvent := &replication.BinlogEvent{RawData: rawData, Header: h, Event: e}
-		binEvent := &replication.BinlogEvent{Header: h, Event: e} // we donnot need raw data
-
-		evs := ParseBinlogEvent(binEvent)
-		for _, ev := range evs {
-			startPos := 0
-			if ev.Type == "query" || ev.Type == "gtid" {
-				startPos = int(h.LogPos - h.EventSize)
-				//fmt.Println(h.Timestamp, h.LogPos-h.EventSize, h.LogPos, db, tb, "sql="+sql, rowCnt, sqlType)
-				//	cfg.StatChan <- BinEventStats{Timestamp: h.Timestamp, Binlog: *binlog, StartPos: h.LogPos - h.EventSize, StopPos: h.LogPos,
-				//		Database: db, Table: tb, QuerySql: sql, RowCnt: rowCnt, QueryType: sqlType}
-			} else {
-				startPos = int(tbMapPos)
-				//fmt.Println(h.Timestamp, t bMapPos, h.LogPos, db, tb, "sql="+sql, rowCnt, sqlType)
-				//	cfg.StatChan <- BinEventStats{Timestamp: h.Timestamp, Binlog: *binlog, StartPos: tbMapPos, StopPos: h.LogPos,
-				//	Database: db, Table: tb, QuerySql: sql, RowCnt: rowCnt, QueryType: sqlType}
-			}
-			switch ev.Type {
-			case "gtid":
-				if currentGtid != "" {
-					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
-					}
-					tx.Size = tx.EndPos - tx.StartPos
-					if f != nil {
-						if !f(tx) {
-							return nil
-						}
-					}
-				}
-				currentGtid = ev.Data
-				tx = Transaction{
-					GTID:      ev.Data,
-					StartPos:  startPos,
-					Timestamp: int64(h.Timestamp),
-					Time:      time.Unix(int64(h.Timestamp), 0),
-				}
-			case "":
-				tx.EndPos = int(h.LogPos)
-				continue
-			case "rowsquery":
-				tx.EndPos = int(h.LogPos)
-				tx.sqlOrigin = append(tx.sqlOrigin, ev.Data)
-			default:
-				tx.EndPos = int(h.LogPos)
-				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),
-					Time:            time.Unix(int64(h.Timestamp), 0),
-					CompressionType: ev.CompressionType,
-				})
-			}
-		}
-	}
-
-}
-
-type BinlogEvent struct {
-	Type            string
-	DB              string
-	TB              string
-	Data            string
-	RowCnt          uint32
-	Rows            [][]interface{}
-	CompressionType string
-}
-
-func ParseBinlogEvent(ev *replication.BinlogEvent) []BinlogEvent {
-	var res []BinlogEvent
-	var sig BinlogEvent
-	switch ev.Header.EventType {
-	case replication.ANONYMOUS_GTID_EVENT:
-		//ge := ev.Event.(*replication.GTIDEvent)
-		sig.Data = "anonymous-gtid-event:1"
-		sig.Type = "gtid"
-	case replication.WRITE_ROWS_EVENTv1,
-		replication.WRITE_ROWS_EVENTv2:
-		wrEvent := ev.Event.(*replication.RowsEvent)
-		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 := ev.Event.(*replication.RowsEvent)
-		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:
-
-		//replication.XID_EVENT,
-		//replication.TABLE_MAP_EVENT:
-
-		wrEvent := ev.Event.(*replication.RowsEvent)
-		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 := ev.Event.(*replication.RowsQueryEvent)
-		sig.Data = string(queryEvent.Query)
-		sig.Type = "rowsquery"
-	case replication.QUERY_EVENT:
-		queryEvent := ev.Event.(*replication.QueryEvent)
-		sig.DB = string(queryEvent.Schema)
-		sig.Data = string(queryEvent.Query)
-		sig.Type = "query"
-
-	case replication.MARIADB_GTID_EVENT:
-		// For global transaction ID, used to start a new transaction event group, instead of the old BEGIN query event, and also to mark stand-alone (ddl).
-		//https://mariadb.com/kb/en/library/gtid_event/
-		sig.Data = "begin"
-		sig.Type = "query"
-
-	case replication.XID_EVENT:
-		// XID_EVENT represents commit。rollback transaction not in binlog
-		sig.Data = "commit"
-		sig.Type = "query"
-	case replication.GTID_EVENT:
-		ge := ev.Event.(*replication.GTIDEvent)
-		gid, err := gtid.Parse(fmt.Sprintf("%s:%d", bytesToUuid(ge.SID), ge.GNO))
-		if err == nil {
-			sig.Data = gid.String()
-		}
-		sig.Type = "gtid"
-	case replication.TRANSACTION_PAYLOAD_EVENT:
-		ge := ev.Event.(*replication.TransactionPayloadEvent)
-		for _, val := range ge.Events {
-			res = append(res, ParseBinlogEvent(val)...)
-		}
-		for idx := range res {
-			if ge.CompressionType == 0 {
-				res[idx].CompressionType = "ZSTD"
-			} else if ge.CompressionType != 255 {
-				res[idx].CompressionType = "UNKNOWN"
-			}
-		}
-		return res
-	}
-	res = append(res, sig)
-	return res
-}
-
-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
-}
-
-func parseBinlogWithFilter(r io.Reader, parse *replication.BinlogParser, filter BinlogFilter, fn func(Transaction) bool) error {
-	var subGtid, inGtid, exGtid *gtid.Gtid
-	var err error
-	var includeMap = make(map[string]bool)
-	var excludeMap = make(map[string]bool)
-	if len(filter.IncludeTables) != 0 {
-		for _, v := range filter.IncludeTables {
-			if len(strings.Split(v, ".")) != 2 {
-				return fmt.Errorf("IncludeTable Name Is Invalid:%s", v)
-			}
-			includeMap[v] = true
-		}
-	} else {
-		includeMap["*.*"] = true
-	}
-	if len(filter.ExcludeTables) != 0 {
-		for _, v := range filter.ExcludeTables {
-			if len(strings.Split(v, ".")) != 2 {
-				return fmt.Errorf("ExcludeTable Name Is Invalid:%s", v)
-			}
-			excludeMap[v] = true
-		}
-	} else {
-		excludeMap["*.*"] = true
-	}
-	if filter.IncludeGtid != "" {
-		inGtid, err = gtid.Parse(filter.IncludeGtid)
-		if err != nil {
-			return err
-		}
-		subGtid = inGtid.Clone()
-	}
-	if filter.ExcludeGtid != "" {
-		exGtid, err = gtid.Parse(filter.ExcludeGtid)
-		if err != nil {
-			return err
-		}
-	}
-	// process: 0, continue: 1, break: 2, EOF: 3
-	var (
-		n            int64
-		tbMapPos     uint32 = 0
-		skipTillNext bool   = false
-	)
-	var tx Transaction
-
-	currentGtid := ""
-	callFn := func(tx Transaction) bool {
-		if fn == nil {
-			return true
-		}
-		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
-		}
-		return fn(tx)
-	}
-	for {
-		headBuf := make([]byte, replication.EventHeaderSize)
-		if _, err = io.ReadFull(r, headBuf); err == io.EOF {
-			if tx.Time.IsZero() {
-				return nil
-			}
-			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 filter.OnlyShowGtid {
-				tx.EndPos = tx.StartPos
-			}
-			tx.Size = tx.EndPos - tx.StartPos
-			callFn(tx)
-			return nil
-		} else if err != nil {
-			return err
-		}
-		var h *replication.EventHeader
-		h, err = parse.ParseHeader(headBuf)
-		if err != nil {
-			return err
-		}
-		//fmt.Printf("parsing %s %d %s\n", *binlog, h.LogPos, GetDatetimeStr(int64(h.Timestamp), int64(0), DATETIME_FORMAT))
-		if h.EventSize <= uint32(replication.EventHeaderSize) {
-			err = fmt.Errorf("invalid event header, event size is %d, too small", h.EventSize)
-			return err
-		}
-
-		var buf bytes.Buffer
-		if n, err = io.CopyN(&buf, r, int64(h.EventSize)-int64(replication.EventHeaderSize)); err != nil {
-			err = fmt.Errorf("get event body err %v, need %d - %d, but got %d", err, h.EventSize, replication.EventHeaderSize, n)
-			return err
-		}
-		if skipTillNext && h.EventType != replication.GTID_EVENT {
-			continue
-		}
-		if filter.OnlyShowGtid && h.EventType != replication.GTID_EVENT {
-			continue
-		}
-		//h.Dump(os.Stdout)
-
-		data := buf.Bytes()
-		var rawData []byte
-		rawData = append(rawData, headBuf...)
-		rawData = append(rawData, data...)
-
-		eventLen := int(h.EventSize) - replication.EventHeaderSize
-
-		if len(data) != eventLen {
-			err = fmt.Errorf("invalid data size %d in event %s, less event length %d", len(data), h.EventType, eventLen)
-			return err
-		}
-
-		var e replication.Event
-		e, err = parse.ParseEvent(h, data, rawData)
-		if err != nil {
-			return err
-		}
-		if h.EventType == replication.TABLE_MAP_EVENT {
-			tbMapPos = h.LogPos - h.EventSize // avoid mysqlbing mask the row event as unknown table row event
-		}
-
-		//e.Dump(os.Stdout)
-
-		//binEvent := &replication.BinlogEvent{RawData: rawData, Header: h, Event: e}
-		binEvent := &replication.BinlogEvent{Header: h, Event: e} // we donnot need raw data
-
-		evs := ParseBinlogEvent(binEvent)
-		for _, ev := range evs {
-			startPos := 0
-			if ev.Type == "query" || ev.Type == "gtid" {
-				startPos = int(h.LogPos - h.EventSize)
-				//fmt.Println(h.Timestamp, h.LogPos-h.EventSize, h.LogPos, db, tb, "sql="+sql, rowCnt, sqlType)
-				//	cfg.StatChan <- BinEventStats{Timestamp: h.Timestamp, Binlog: *binlog, StartPos: h.LogPos - h.EventSize, StopPos: h.LogPos,
-				//		Database: db, Table: tb, QuerySql: sql, RowCnt: rowCnt, QueryType: sqlType}
-			} else {
-				startPos = int(tbMapPos)
-				//fmt.Println(h.Timestamp, t bMapPos, h.LogPos, db, tb, "sql="+sql, rowCnt, sqlType)
-				//	cfg.StatChan <- BinEventStats{Timestamp: h.Timestamp, Binlog: *binlog, StartPos: tbMapPos, StopPos: h.LogPos,
-				//	Database: db, Table: tb, QuerySql: sql, RowCnt: rowCnt, QueryType: sqlType}
-			}
-			if filter.EndPos != 0 && startPos > filter.EndPos {
-				skipTillNext = true
-				continue
-			}
-			if filter.StartPos != 0 && startPos < filter.EndPos {
-				skipTillNext = true
-				continue
-			}
-			switch ev.Type {
-			case "gtid":
-				if skipTillNext {
-					skipTillNext = false
-				}
-				if currentGtid != "" {
-					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 filter.OnlyShowGtid {
-						tx.EndPos = startPos - 1
-					}
-					tx.Size = tx.EndPos - tx.StartPos
-					if !callFn(tx) {
-						return nil
-					}
-					if subGtid != nil {
-						subGtid.Sub(tx.GTID)
-						if subGtid.EventCount() == 0 {
-							return nil
-						}
-					}
-					tx = Transaction{}
-				}
-				currentGtid = ev.Data
-				if inGtid != nil {
-					if c, _ := inGtid.Contain(ev.Data); !c {
-						currentGtid = ""
-						skipTillNext = true
-						continue
-					}
-				}
-				if exGtid != nil {
-					if c, _ := exGtid.Contain(ev.Data); c {
-						currentGtid = ""
-						skipTillNext = true
-						continue
-					}
-				}
-				tx = Transaction{
-					GTID:      ev.Data,
-					StartPos:  startPos,
-					Timestamp: int64(h.Timestamp),
-					Time:      time.Unix(int64(h.Timestamp), 0),
-				}
-			case "":
-				tx.EndPos = int(h.LogPos)
-				continue
-			case "rowsquery":
-				tx.EndPos = int(h.LogPos)
-				tx.sqlOrigin = append(tx.sqlOrigin, ev.Data)
-			default:
-				tx.EndPos = int(h.LogPos)
-				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),
-					Time:            time.Unix(int64(h.Timestamp), 0),
-					CompressionType: ev.CompressionType,
-				})
-			}
-		}
-	}
-}
-
-func ParseBinlogWithFilter(path string, pos int64, filter BinlogFilter, fx func(Transaction) bool) error {
-	defer func() {
-		recover()
-	}()
-	if !staros.Exists(path) {
-		return os.ErrNotExist
-	}
-	f, err := os.Open(path)
-	if f != nil {
-		defer f.Close()
-	}
-	if err != nil {
-		return err
-	}
-	parse := replication.NewBinlogParser()
-	parse.SetParseTime(false)
-	parse.SetUseDecimal(false)
-	seekZore := func() error {
-		fileTypeBytes := int64(4)
-		b := make([]byte, fileTypeBytes)
-		// 读取binlog头
-		if _, err = f.Read(b); err != nil {
-			return err
-		} else if !bytes.Equal(b, replication.BinLogFileHeader) {
-			//不是binlog格式
-			return err
-		}
-		// must not seek to other position, otherwise the program may panic because formatevent, table map event is skipped
-		if _, err = f.Seek(fileTypeBytes, os.SEEK_SET); err != nil {
-			return err
-		}
-		return nil
-	}
-	if pos != 0 {
-		if err = seekZore(); err != nil {
-			return err
-		}
-		for {
-			headBuf := make([]byte, replication.EventHeaderSize)
-			if _, err = io.ReadFull(f, headBuf); err != nil {
-				return err
-			}
-			var h *replication.EventHeader
-			h, err = parse.ParseHeader(headBuf)
-			if err != nil {
-				return err
-			}
-			if h.EventSize <= uint32(replication.EventHeaderSize) {
-				err = fmt.Errorf("invalid event header, event size is %d, too small", h.EventSize)
-				return err
-			}
-			var buf bytes.Buffer
-			if n, err := io.CopyN(&buf, f, int64(h.EventSize)-int64(replication.EventHeaderSize)); err != nil {
-				err = fmt.Errorf("get event body err %v, need %d - %d, but got %d", err, h.EventSize, replication.EventHeaderSize, n)
-				return err
-			}
-			data := buf.Bytes()
-			var rawData []byte
-			rawData = append(rawData, headBuf...)
-			rawData = append(rawData, data...)
-			eventLen := int(h.EventSize) - replication.EventHeaderSize
-			if len(data) != eventLen {
-				err = fmt.Errorf("invalid data size %d in event %s, less event length %d", len(data), h.EventType, eventLen)
-				return err
-			}
-			_, err = parse.ParseEvent(h, data, rawData)
-			if err != nil {
-				return err
-			}
-			if h.EventType == replication.FORMAT_DESCRIPTION_EVENT || h.EventType == replication.GTID_EVENT {
-				break
-			}
-		}
-		if _, err = f.Seek(pos, 0); err != nil {
-			return err
-		}
-	}
-
-	if pos == 0 {
-		if err = seekZore(); err != nil {
-			return err
-		}
-	}
-	return parseBinlogWithFilter(f, parse, filter, fx)
-}
+// Parsing implementation is split across:
+// - parse_types.go
+// - parse_io.go
+// - parse_event_convert.go
+// - parse_stream.go
+// - parse_filter.go

parse_event_convert.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)
+	}
+}

parse_event_convert_payload_test.go

@@ -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)
+	}
+}

parse_event_convert_unsigned_test.go

@@ -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])
+	}
+}

parse_filter.go

@@ -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
+}

parse_filter_table_test.go

@@ -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])
+	}
+}

parse_io.go

@@ -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
+}

parse_io_test.go

@@ -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)
+	}
+}

parse_options_test.go

@@ -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)
+	}
+}

parse_stream.go

@@ -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)
+		}
+	}
+}

parse_types.go

@@ -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
+}

summary.go

@@ -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{}
+}

summary_test.go

@@ -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)
+	}
+}