feat(parquet): voids_id column for tombstone round-trip

Tombstone events lost VoidsID through the parquet layer, breaking
read-side attestation voiding. Optional column; old files read fine.

Author: zer0stars

parquet/decoder.go

@@ -134,6 +134,7 @@ func convertRow(row *ParquetRow) (cloudevent.StoredEvent, error) {
 			CloudEventHeader: header,
 		},
 		DataIndexKey: row.DataIndexKey,
+		VoidsID:      row.VoidsID,
 	}
 
 	if len(row.DataBase64) > 0 {

parquet/encoder.go

@@ -204,6 +204,7 @@ func convertEvent(event *cloudevent.StoredEvent) (ParquetRow, error) {
 		Producer:        event.Producer,
 		Extras:          string(extrasJSON),
 		DataIndexKey:    event.DataIndexKey,
+		VoidsID:         event.VoidsID,
 	}
 
 	if event.DataBase64 != "" {

parquet/parquet_row.go

@@ -15,4 +15,5 @@ type ParquetRow struct {
 	Data            *string   `parquet:"data,optional"`
 	DataBase64      []byte    `parquet:"data_base64,optional"`
 	DataIndexKey    string    `parquet:"data_index_key,optional"`
+	VoidsID         string    `parquet:"voids_id,optional"`
 }

parquet/parquet_test.go

@@ -822,3 +822,19 @@ func TestEncode_AllNewOptionsCombined(t *testing.T) {
 	assert.Equal(t, "e-1", got[0].ID)
 	assert.Equal(t, "e-2", got[1].ID)
 }
+
+func TestDecode_VoidsIDRoundTrip(t *testing.T) {
+	t.Parallel()
+	ev := cloudevent.StoredEvent{
+		RawEvent: makeEvent("evt-tomb", json.RawMessage(`{"reason":"superseded"}`)),
+		VoidsID:  "evt-voided",
+	}
+	var buf bytes.Buffer
+	_, err := Encode(&buf, []cloudevent.StoredEvent{ev}, "key")
+	require.NoError(t, err)
+
+	events, err := Decode(bytes.NewReader(buf.Bytes()), int64(buf.Len()))
+	require.NoError(t, err)
+	require.Len(t, events, 1)
+	assert.Equal(t, "evt-voided", events[0].VoidsID)
+}