Add data_index_key (#53)

We wanted to store the data from large documents (per some configurable threshold) in separate S3 objects so that they can be downloaded easily through presigned S3 links and not balloon the size of Parquet files. Since DIMO-Network/dis#254 we've been storing these large events as standalone JSON objects, including the headers. But clients do not enjoy unwrapping the data inside the event and parsing the base64.

To accommodate this, we introduce a new ClickHouse column `data_index_key`. Unlike the existing `index_key`, this column is also written to the Parquet files (morally, we want to be able to reconstruct a working system from S3 alone). In Go we represent this in a new `StoredEvent` type, rather than adding to `CloudEventHeader`, for two reasons:

1. Security. With this choice it's easier to ensure that client input is never placed into this field. If we failed at this, a client might be able to gain access to files that he should not be able to see.
2. Separation. Many users of the cloudevent package don't care about ClickHouse storage at all. The struct currently doesn't bother them with an `IndexKey` field, and it would be nice to maintain this separation.

Old clients that assume the old table structures should still work, but they should upgrade soon.

Author: elffjs

README.md

@@ -110,7 +110,14 @@ A CloudEvent is uniquely identified by the combination of the following fields:
 - `Source`
 - `ID`
 
-This combination forms the "index key" for the event and is used for deduplication and retrieval purposes.
+This combination is the event's logical key (see `CloudEventHeader.Key()`) and is used for deduplication.
+
+### Storage keys
+
+These are storage pointers, set server-side by the ingestion pipeline. They are **not** part of the CloudEvent wire format and must never be set from producer-supplied input.
+
+- `index_key`: pointer to the backing object holding this event's row. For events grouped into a Parquet bundle, the format is `<bundleObjectKey>#<rowOffset>` (see `parquet.ParseIndexKey`). For singleton events, it's the standalone object's key (see `clickhouse.CloudEventToObjectKey`).
+- `data_index_key`: pointer to the external object holding this event's payload, when the payload is stored out-of-band (e.g. a large image). Empty when the payload is inline in the row's `data` / `data_base64`. Carried on `cloudevent.StoredEvent.DataIndexKey` at write time and stored as a column in both ClickHouse and the Parquet bundle.
 
 ### DIMO Event Types
 

clickhouse/clickhouse.go

@@ -33,6 +33,8 @@ const (
 	ExtrasColumn = "extras"
 	// IndexKeyColumn is the name of the index name column in Clickhouse.
 	IndexKeyColumn = "index_key"
+	// DataIndexKeyColumn is the name of the data index name column in Clickhouse.
+	DataIndexKeyColumn = "data_index_key"
 
 	// InsertStmt is the SQL statement for inserting a row into Clickhouse.
 	InsertStmt = "INSERT INTO " + TableName + " (" +
@@ -45,22 +47,37 @@ const (
 		DataContentTypeColumn + ", " +
 		DataVersionColumn + ", " +
 		ExtrasColumn + ", " +
-		IndexKeyColumn +
-		") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+		IndexKeyColumn + ", " +
+		DataIndexKeyColumn +
+		") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
 
 	// hexChars contains the characters used for hex representation
 	hexChars = "0123456789abcdef"
 )
 
 // CloudEventToSlice converts a CloudEvent to an array of any for Clickhouse insertion.
 // The order of the elements in the array match the order of the columns in the table.
+// index_key is computed from the event's headers.
 func CloudEventToSlice(event *cloudevent.CloudEventHeader) []any {
 	return CloudEventToSliceWithKey(event, CloudEventToObjectKey(event))
 }
 
 // CloudEventToSliceWithKey converts a CloudEvent to an array of any for Clickhouse insertion.
 // The order of the elements in the array match the order of the columns in the table.
+// This variant allows the caller to specify a value for index_key.
 func CloudEventToSliceWithKey(event *cloudevent.CloudEventHeader, key string) []any {
+	return cloudEventToSlice(event, key, "")
+}
+
+// StoredEventToSlice converts a StoredEvent to an array of any for Clickhouse
+// insertion, populating both index_key (caller-supplied) and data_index_key
+// (carried on the wrapper). The order of the elements matches the column
+// order in the table.
+func StoredEventToSlice(stored *cloudevent.StoredEvent, indexKey string) []any {
+	return cloudEventToSlice(&stored.CloudEventHeader, indexKey, stored.DataIndexKey)
+}
+
+func cloudEventToSlice(event *cloudevent.CloudEventHeader, indexKey, dataIndexKey string) []any {
 	// Add non-column fields to extras
 	extras := cloudevent.AddNonColumnFieldsToExtras(event)
 
@@ -80,7 +97,8 @@ func CloudEventToSliceWithKey(event *cloudevent.CloudEventHeader, key string) []
 		event.DataContentType,
 		event.DataVersion,
 		string(jsonExtra),
-		key,
+		indexKey,
+		dataIndexKey,
 	}
 }
 
@@ -90,11 +108,11 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) {
 	if err := json.Unmarshal(jsonArray, &rawSlice); err != nil {
 		return nil, fmt.Errorf("failed to unmarshal cloud event slice: %w", err)
 	}
-	if len(rawSlice) != 10 {
+	if len(rawSlice) != 11 {
 		return nil, fmt.Errorf("invalid cloud event slice length: %d", len(rawSlice))
 	}
 
-	// Column order: subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey
+	// Column order: subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey
 	var (
 		subject         string
 		timestamp       time.Time
@@ -106,6 +124,7 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) {
 		dataVersion     string
 		extras          string
 		indexKey        string
+		dataIndexKey    string
 	)
 	unmarshal := func(i int, name string, ptr any) error {
 		if err := json.Unmarshal(rawSlice[i], ptr); err != nil {
@@ -143,7 +162,10 @@ func UnmarshalCloudEventSlice(jsonArray []byte) ([]any, error) {
 	if err := unmarshal(9, "index key", &indexKey); err != nil {
 		return nil, err
 	}
-	return []any{subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey}, nil
+	if err := unmarshal(10, "data index key", &dataIndexKey); err != nil {
+		return nil, err
+	}
+	return []any{subject, timestamp, eventType, id, source, producer, dataContentType, dataVersion, extras, indexKey, dataIndexKey}, nil
 }
 
 // CloudEventToObjectKey generates a unique key for storing cloud events.

clickhouse/clickhouse_test.go

@@ -33,7 +33,7 @@ func TestCloudEventToSlice(t *testing.T) {
 
 	// Test CloudEventToSlice
 	slice := CloudEventToSlice(event)
-	require.Len(t, slice, 10)
+	require.Len(t, slice, 11)
 
 	// Verify the order and values of the slice
 	assert.Equal(t, event.Subject, slice[0])
@@ -57,6 +57,9 @@ func TestCloudEventToSlice(t *testing.T) {
 	// Verify index key
 	expectedKey := CloudEventToObjectKey(event)
 	assert.Equal(t, expectedKey, slice[9])
+
+	// data_index_key is empty for CloudEventToSlice (payload is not external)
+	assert.Equal(t, "", slice[10])
 }
 
 func TestCloudEventToSliceWithKey(t *testing.T) {
@@ -82,7 +85,7 @@ func TestCloudEventToSliceWithKey(t *testing.T) {
 
 	customKey := "custom-key"
 	slice := CloudEventToSliceWithKey(event, customKey)
-	require.Len(t, slice, 10)
+	require.Len(t, slice, 11)
 
 	// Verify the order and values of the slice
 	assert.Equal(t, event.Subject, slice[0])
@@ -105,6 +108,38 @@ func TestCloudEventToSliceWithKey(t *testing.T) {
 
 	// Verify custom key is used
 	assert.Equal(t, customKey, slice[9])
+
+	// data_index_key is empty for CloudEventToSliceWithKey (payload is not external)
+	assert.Equal(t, "", slice[10])
+}
+
+func TestStoredEventToSlice(t *testing.T) {
+	t.Parallel()
+
+	now := time.Now().UTC().Truncate(time.Millisecond)
+	stored := &cloudevent.StoredEvent{
+		RawEvent: cloudevent.RawEvent{
+			CloudEventHeader: cloudevent.CloudEventHeader{
+				ID:              "test-id",
+				Source:          "test-source",
+				Producer:        "test-producer",
+				SpecVersion:     "1.0",
+				Subject:         "test-subject",
+				Time:            now,
+				Type:            "test.type",
+				DataContentType: "image/jpeg",
+				DataVersion:     "v1",
+			},
+		},
+		DataIndexKey: "payloads/abc123.jpg",
+	}
+
+	slice := StoredEventToSlice(stored, "bundle/key#7")
+	require.Len(t, slice, 11)
+
+	assert.Equal(t, stored.Subject, slice[0])
+	assert.Equal(t, "bundle/key#7", slice[9])
+	assert.Equal(t, "payloads/abc123.jpg", slice[10])
 }
 
 func TestUnmarshalCloudEventSlice(t *testing.T) {
@@ -122,6 +157,7 @@ func TestUnmarshalCloudEventSlice(t *testing.T) {
 		"v1",
 		`{"extra1":"value1","extra2":123}`,
 		"test-key",
+		"test-data-key",
 	}
 
 	// Marshal the slice to JSON

clickhouse/migrations/00007_add_data_index_key_migration.sql

@@ -0,0 +1,9 @@
+-- +goose Up
+-- +goose StatementBegin
+ALTER TABLE cloud_event ADD COLUMN data_index_key String COMMENT 'Key of the object holding the data payload. Empty for small events that are entirely stored in the object referenced by index_key.' AFTER index_key;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE cloud_event DROP COLUMN data_index_key;
+-- +goose StatementEnd

clickhouse/migrations/migrations_test.go

@@ -59,6 +59,7 @@ func TestMigration(t *testing.T) {
 		localch.DataVersionColumn,
 		localch.ExtrasColumn,
 		localch.IndexKeyColumn,
+		localch.DataIndexKeyColumn,
 	}
 	assert.ElementsMatch(t, expectedCols, cols, "Columns do not match")
 

parquet/decoder.go

@@ -35,19 +35,19 @@ func (pr *Reader) NumRows() int64 {
 
 // SeekToRow retrieves a single row by index.
 // The rowIndex must be in the range [0, NumRows()).
-func (pr *Reader) SeekToRow(rowIndex int64) (cloudevent.RawEvent, error) {
+func (pr *Reader) SeekToRow(rowIndex int64) (cloudevent.StoredEvent, error) {
 	if rowIndex < 0 || rowIndex >= pr.reader.NumRows() {
-		return cloudevent.RawEvent{}, fmt.Errorf("row index %d out of range [0, %d)", rowIndex, pr.reader.NumRows())
+		return cloudevent.StoredEvent{}, fmt.Errorf("row index %d out of range [0, %d)", rowIndex, pr.reader.NumRows())
 	}
 
 	if err := pr.reader.SeekToRow(rowIndex); err != nil {
-		return cloudevent.RawEvent{}, fmt.Errorf("seeking to row %d: %w", rowIndex, err)
+		return cloudevent.StoredEvent{}, fmt.Errorf("seeking to row %d: %w", rowIndex, err)
 	}
 
 	var rows [1]ParquetRow
 	_, err := pr.reader.Read(rows[:])
 	if err != nil && err != io.EOF {
-		return cloudevent.RawEvent{}, fmt.Errorf("reading parquet row: %w", err)
+		return cloudevent.StoredEvent{}, fmt.Errorf("reading parquet row: %w", err)
 	}
 
 	return convertRow(&rows[0])
@@ -60,7 +60,7 @@ func (pr *Reader) Close() error {
 
 // Decode reads a parquet file from r and returns the decoded CloudEvents.
 // The size parameter must be the total size of the parquet data in bytes.
-func Decode(r io.ReaderAt, size int64) ([]cloudevent.RawEvent, error) {
+func Decode(r io.ReaderAt, size int64) ([]cloudevent.StoredEvent, error) {
 	pr, err := OpenReader(r, size)
 	if err != nil {
 		return nil, err
@@ -78,7 +78,7 @@ func Decode(r io.ReaderAt, size int64) ([]cloudevent.RawEvent, error) {
 		return nil, fmt.Errorf("reading parquet rows: %w", err)
 	}
 
-	events := make([]cloudevent.RawEvent, len(rows))
+	events := make([]cloudevent.StoredEvent, len(rows))
 	for i := range rows {
 		event, err := convertRow(&rows[i])
 		if err != nil {
@@ -94,23 +94,23 @@ func Decode(r io.ReaderAt, size int64) ([]cloudevent.RawEvent, error) {
 // The rowIndex must be in the range [0, numRows).
 // For multiple seeks on the same file, use OpenReader instead to avoid
 // re-parsing the parquet footer on each call.
-func SeekToRow(r io.ReaderAt, size int64, rowIndex int64) (cloudevent.RawEvent, error) {
+func SeekToRow(r io.ReaderAt, size int64, rowIndex int64) (cloudevent.StoredEvent, error) {
 	pr, err := OpenReader(r, size)
 	if err != nil {
-		return cloudevent.RawEvent{}, err
+		return cloudevent.StoredEvent{}, err
 	}
 	defer func() { _ = pr.Close() }()
 
 	return pr.SeekToRow(rowIndex)
 }
 
-// convertRow transforms a single ParquetRow back into a RawEvent.
+// convertRow transforms a single ParquetRow back into a StoredEvent.
 // This is the inverse of convertEvent in encoder.go.
-func convertRow(row *ParquetRow) (cloudevent.RawEvent, error) {
+func convertRow(row *ParquetRow) (cloudevent.StoredEvent, error) {
 	var extras map[string]any
 	if row.Extras != "" && row.Extras != "{}" {
 		if err := json.Unmarshal([]byte(row.Extras), &extras); err != nil {
-			return cloudevent.RawEvent{}, fmt.Errorf("unmarshaling extras: %w", err)
+			return cloudevent.StoredEvent{}, fmt.Errorf("unmarshaling extras: %w", err)
 		}
 	}
 
@@ -129,8 +129,11 @@ func convertRow(row *ParquetRow) (cloudevent.RawEvent, error) {
 
 	cloudevent.RestoreNonColumnFields(&header)
 
-	event := cloudevent.RawEvent{
-		CloudEventHeader: header,
+	event := cloudevent.StoredEvent{
+		RawEvent: cloudevent.RawEvent{
+			CloudEventHeader: header,
+		},
+		DataIndexKey: row.DataIndexKey,
 	}
 
 	if len(row.DataBase64) > 0 {

parquet/encoder.go

@@ -53,7 +53,12 @@ func WithWriteBufferSize(n int) Option {
 // Encode writes events as Snappy-compressed Parquet to w. Each event is assigned
 // an index key in the format "objectKey#rowOffset". The returned map contains
 // the event index to index key mapping.
-func Encode(w io.Writer, events []cloudevent.RawEvent, objectKey string, opts ...Option) (map[int]string, error) {
+//
+// Each StoredEvent's DataIndexKey is written into the row so that a reader
+// holding a bundle alone can locate any externally-stored payload without
+// consulting ClickHouse. Leave DataIndexKey empty when the payload is inline
+// in Data / DataBase64.
+func Encode(w io.Writer, events []cloudevent.StoredEvent, objectKey string, opts ...Option) (map[int]string, error) {
 	cfg := EncoderConfig{
 		MaxRowsPerRowGroup: 10000,
 	}
@@ -101,8 +106,8 @@ func Encode(w io.Writer, events []cloudevent.RawEvent, objectKey string, opts ..
 	return indexKeys, nil
 }
 
-// convertEvent transforms a single CloudEvent into a ParquetRow.
-func convertEvent(event *cloudevent.RawEvent) (ParquetRow, error) {
+// convertEvent transforms a single StoredEvent into a ParquetRow.
+func convertEvent(event *cloudevent.StoredEvent) (ParquetRow, error) {
 	extras := cloudevent.AddNonColumnFieldsToExtras(&event.CloudEventHeader)
 
 	var extrasJSON []byte
@@ -126,6 +131,7 @@ func convertEvent(event *cloudevent.RawEvent) (ParquetRow, error) {
 		DataVersion:     event.DataVersion,
 		Producer:        event.Producer,
 		Extras:          string(extrasJSON),
+		DataIndexKey:    event.DataIndexKey,
 	}
 
 	if event.DataBase64 != "" {

parquet/parquet_row.go

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