From 28b8bdb5772e096c58c9bdb7ca9da52f43fe9eb5 Mon Sep 17 00:00:00 2001
From: Ilia Demianenko <ilia.demianenko@clickhouse.com>
Date: Thu, 1 Oct 2026 11:24:35 -0600
Subject: [PATCH] Accept InsertMessage with zero-column tuples

PostgreSQL emits zero-column tuples for inserts into tables with no
replicated columns: tables created with no columns, tables whose columns
have all been dropped, and (before PG 18) tables with only generated
columns. The payload is then 7 bytes (relation ID, 'N', column count 0),
but InsertMessage.Decode required at least 8, so decoding failed with
"InsertMessage must have 8 bytes, got 7 bytes" and a replication stream
could not progress past such a change. The in-stream InsertMessageV2
check had the same off-by-one (12 instead of 11).
---
 message.go        |  4 ++--
 messageV2.go      |  4 ++--
 messageV2_test.go | 23 +++++++++++++++++++++++
 message_test.go   | 29 +++++++++++++++++++++++++++++
 4 files changed, 56 insertions(+), 4 deletions(-)

diff --git a/message.go b/message.go
index 6315532..2a56c8d 100644
--- a/message.go
+++ b/message.go
@@ -438,8 +438,8 @@ type InsertMessage struct {
 
 // Decode decodes to message from src.
 func (m *InsertMessage) Decode(src []byte) error {
-	if len(src) < 8 {
-		return m.lengthError("InsertMessage", 8, len(src))
+	if len(src) < 7 {
+		return m.lengthError("InsertMessage", 7, len(src))
 	}
 
 	var low, used int
diff --git a/messageV2.go b/messageV2.go
index 990e46f..43c9e31 100644
--- a/messageV2.go
+++ b/messageV2.go
@@ -237,8 +237,8 @@ func (m *InsertMessageV2) DecodeV2(src []byte, inStream bool) (err error) {
 		return m.InsertMessage.Decode(src)
 	}
 
-	if len(src) < 12 {
-		return m.lengthError("InsertMessageV2", 12, len(src))
+	if len(src) < 11 {
+		return m.lengthError("InsertMessageV2", 11, len(src))
 	}
 
 	src = readXidAndAdvance(src, &m.InStreamMessageV2WithXid, inStream)
diff --git a/messageV2_test.go b/messageV2_test.go
index 9f98fbc..15a0892 100644
--- a/messageV2_test.go
+++ b/messageV2_test.go
@@ -271,6 +271,29 @@ func (s *insertMessageV2Suite) TestNoStream() {
 	s.Equal(expected, &insertMsg.InsertMessage)
 }
 
+func (s *insertMessageV2Suite) TestNoColumns() {
+	msg, expected := s.createInsertTestDataNoColumns()
+	msgV2, xid := s.insertXid(msg)
+
+	m, err := ParseV2(msgV2, true)
+	s.NoError(err)
+	insertMsg, ok := m.(*InsertMessageV2)
+	s.True(ok)
+	s.Equal(xid, insertMsg.Xid)
+	s.Equal(expected, &insertMsg.InsertMessage)
+}
+
+func (s *insertMessageV2Suite) TestNoColumnsNoStream() {
+	msg, expected := s.createInsertTestDataNoColumns()
+
+	m, err := ParseV2(msg, false)
+	s.NoError(err)
+	insertMsg, ok := m.(*InsertMessageV2)
+	s.True(ok)
+	s.Equal(uint32(0), insertMsg.Xid)
+	s.Equal(expected, &insertMsg.InsertMessage)
+}
+
 func TestUpdateMessageV2Suite(t *testing.T) {
 	suite.Run(t, new(updateMessageV2Suite))
 }
diff --git a/message_test.go b/message_test.go
index 069117a..9c48acc 100644
--- a/message_test.go
+++ b/message_test.go
@@ -285,6 +285,24 @@ func (s *messageSuite) createInsertTestData() ([]byte, *InsertMessage) {
 	return msg, expected
 }
 
+func (s *messageSuite) createInsertTestDataNoColumns() ([]byte, *InsertMessage) {
+	relationID := s.newRelationID()
+
+	msg := make([]byte, 1+4+1+2)
+	msg[0] = 'I'
+	bigEndian.PutUint32(msg[1:], relationID)
+	msg[5] = 'N'
+	bigEndian.PutUint16(msg[6:], 0)
+
+	expected := &InsertMessage{
+		RelationID: relationID,
+		Tuple:      &TupleData{},
+	}
+	expected.msgType = 'I'
+
+	return msg, expected
+}
+
 func (s *messageSuite) createUpdateTestDataTypeK() ([]byte, *UpdateMessage) {
 	relationID := s.newRelationID()
 
@@ -736,6 +754,17 @@ func (s *insertMessageSuite) Test() {
 	s.Equal(expected, insertMsg)
 }
 
+func (s *insertMessageSuite) TestNoColumns() {
+	msg, expected := s.createInsertTestDataNoColumns()
+
+	m, err := Parse(msg)
+	s.NoError(err)
+	insertMsg, ok := m.(*InsertMessage)
+	s.True(ok)
+
+	s.Equal(expected, insertMsg)
+}
+
 func TestUpdateMessageSuite(t *testing.T) {
 	suite.Run(t, new(updateMessageSuite))
 }