postgres: page sum impl

format/postgres/common/checksum.go

@@ -0,0 +1,67 @@
+package common
+
+import "encoding/binary"
+
+const (
+	nSums        = 32
+	fnvPrime     = 16777619
+	nSumsSize    = 4 * nSums
+	mainBlockLen = 8192 / nSumsSize
+)
+
+var (
+	checksumBaseOffsets = [nSums]uint32{
+		0x5B1F36E9, 0xB8525960, 0x02AB50AA, 0x1DE66D2A,
+		0x79FF467A, 0x9BB9F8A3, 0x217E7CD2, 0x83E13D2C,
+		0xF8D4474F, 0xE39EB970, 0x42C6AE16, 0x993216FA,
+		0x7B093B5D, 0x98DAFF3C, 0xF718902A, 0x0B1C9CDB,
+		0xE58F764B, 0x187636BC, 0x5D7B3BB1, 0xE73DE7DE,
+		0x92BEC979, 0xCCA6C0B2, 0x304A0979, 0x85AA43D4,
+		0x783125BB, 0x6CA8EAA2, 0xE407EAC6, 0x4B5CFC3E,
+		0x9FBF8C76, 0x15CA20BE, 0xF2CA9FD3, 0x959BD756,
+	}
+)
+
+func checksumComp(checksum uint32, value uint32) uint32 {
+	__tmp := checksum ^ value
+	checksum = __tmp*fnvPrime ^ (__tmp >> 17)
+	return checksum
+}
+
+func pgChecksumBlock(page []byte) uint32 {
+	sums := [nSums]uint32{0}
+	for i := 0; i < nSums; i++ {
+		sums[i] = checksumBaseOffsets[i]
+	}
+	result := uint32(0)
+
+	for i := 0; i < mainBlockLen; i++ {
+		for j := 0; j < nSums; j++ {
+			v2 := binary.LittleEndian.Uint32(page[i*nSumsSize+j*4:])
+			sums[j] = checksumComp(sums[j], v2)
+		}
+	}
+
+	for i := 0; i < 2; i++ {
+		for j := 0; j < nSums; j++ {
+			sums[j] = checksumComp(sums[j], 0)
+		}
+	}
+
+	for i := 0; i < nSums; i++ {
+		result = result ^ sums[i]
+	}
+
+	return result
+}
+
+func CheckSumBlock(page []byte, blockNumber uint32) uint16 {
+	// set pd_checksum to zero
+	page[8] = 0
+	page[9] = 0
+
+	sum := pgChecksumBlock(page)
+	sum1 := sum ^ blockNumber
+	sum2 := uint16((sum1 % 65535) + 1)
+	return sum2
+}

format/postgres/flavours/pgproee14/ee14/pgheap.go

@@ -22,7 +22,7 @@ func DecodePageHeaderData(page *common14.HeapPage, d *decode.D) {
 		d.FieldU32("xlogid", common.HexMapper)
 		d.FieldU32("xrecoff", common.HexMapper)
 	})
-	d.FieldU16("pd_checksum")
+	page.PdChecksum = uint16(d.FieldU16("pd_checksum"))
 	d.FieldU16("pd_flags")
 	page.PdLower = uint16(d.FieldU16("pd_lower"))
 	page.PdUpper = uint16(d.FieldU16("pd_upper"))

format/postgres/flavours/postgres14/common14/heap_page.go

@@ -23,6 +23,7 @@ type ItemID struct {
 
 type HeapPage struct {
 	// PageHeaderData fields
+	PdChecksum        uint16
 	PdLower           uint16
 	PdUpper           uint16
 	PdSpecial         uint16
@@ -48,7 +49,7 @@ func DecodePageHeader(page *HeapPage, d *decode.D) {
 		d.FieldU32("xlogid", common.HexMapper)
 		d.FieldU32("xrecoff", common.HexMapper)
 	})
-	d.FieldU16("pd_checksum")
+	page.PdChecksum = uint16(d.FieldU16("pd_checksum"))
 	d.FieldU16("pd_flags")
 	page.PdLower = uint16(d.FieldU16("pd_lower"))
 	page.PdUpper = uint16(d.FieldU16("pd_upper"))

format/postgres/flavours/postgres14/common14/pg_heap.go

@@ -2,7 +2,6 @@ package common14
 
 import (
 	"fmt"
-
 	"github.com/wader/fq/format/postgres/common"
 
 	"github.com/wader/fq/pkg/decode"
@@ -145,14 +144,16 @@ func DecodeHeap(heap *Heap, d *decode.D) any {
 }
 
 func decodeHeapPages(heap *Heap, d *decode.D) {
+	blockNumber := uint32(0)
 	for {
 		if end, _ := d.TryEnd(); end {
 			return
 		}
 
 		d.FieldStruct("page", func(d *decode.D) {
-			decodeHeapPage(heap, d)
+			decodeHeapPage(heap, d, blockNumber)
 		})
+		blockNumber++
 
 		// end of Page
 		endLen := uint64(d.Pos() / 8)
@@ -161,7 +162,7 @@ func decodeHeapPages(heap *Heap, d *decode.D) {
 	}
 }
 
-func decodeHeapPage(heap *Heap, d *decode.D) {
+func decodeHeapPage(heap *Heap, d *decode.D, blockNumber uint32) {
 	page := &HeapPage{}
 	if heap.Page != nil {
 		// use prev page
@@ -171,8 +172,14 @@ func decodeHeapPage(heap *Heap, d *decode.D) {
 	heap.Page = page
 	heap.Special = &PageSpecial{}
 
+	checkSum := calcCheckSum(d, heap.PageSize, blockNumber)
+
 	d.FieldStruct("page_header", func(d *decode.D) {
 		heap.DecodePageHeaderData(page, d)
+
+		d.FieldValueU("pd_checksum_check", uint64(checkSum))
+		sumEqual := page.PdChecksum == checkSum
+		d.FieldValueBool("pd_checksum_check_equal", sumEqual)
 	})
 
 	DecodeItemIds(page, d)
@@ -187,6 +194,19 @@ func decodeHeapPage(heap *Heap, d *decode.D) {
 	})
 }
 
+func calcCheckSum(d *decode.D, pageSize uint64, blockNumber uint32) uint16 {
+	pos0 := d.Pos()
+	pageBuffer := make([]byte, pageSize)
+	rdrPage := d.RawLen(int64(pageSize * 8))
+	_, err := rdrPage.ReadBits(pageBuffer, int64(pageSize*8))
+	if err != nil {
+		d.Fatalf("can't read page, err = %v\n", err)
+	}
+	sum2 := common.CheckSumBlock(pageBuffer, blockNumber)
+	d.SeekAbs(pos0)
+	return sum2
+}
+
 func decodeTuples(heap *Heap, d *decode.D) {
 	page := heap.Page
 	for i := 0; i < len(page.ItemIds); i++ {