Merge pull request #3082 from redpanda-data/avro-meta

CHANGELOG.md

@@ -3,6 +3,12 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.45.0 - TBD
+
+### Added
+
+- `avro` scanner now emits metadata for the Avro schema it used along with the schema fingerprint (@rockwotj)
+
 ## 4.44.0 - 2024-12-13
 
 ### Added

docs/modules/components/pages/scanners/avro.adoc

@@ -65,6 +65,8 @@ For example, the union schema `["null","string","Foo"]`, where `Foo` is a record
 
 However, it is possible to instead create documents in https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull[standard/raw JSON format^] by setting the field <<avro_raw_json,`avro_raw_json`>> to `true`.
 
+This scanner also emits the canonical Avro schema as `@avro_schema` metadata, along with the schema's fingerprint available via `@avro_schema_fingerprint`.
+
 
 == Fields
 

internal/impl/avro/scanner.go

@@ -48,6 +48,8 @@ For example, the union schema ` + "`[\"null\",\"string\",\"Foo\"]`, where `Foo`"
 - a ` + "`Foo` instance as `{\"Foo\": {...}}`, where `{...}` indicates the JSON encoding of a `Foo`" + ` instance.
 
 However, it is possible to instead create documents in https://pkg.go.dev/github.com/linkedin/goavro/v2#NewCodecForStandardJSONFull[standard/raw JSON format^] by setting the field ` + "<<avro_raw_json,`avro_raw_json`>> to `true`" + `.
+
+This scanner also emits the canonical Avro schema as ` + "`@avro_schema`" + ` metadata, along with the schema's fingerprint available via ` + "`@avro_schema_fingerprint`" + `.
 `).
 		Fields(
 			service.NewBoolField(sFieldRawJSON).
@@ -133,7 +135,10 @@ func (c *avroScanner) NextBatch(ctx context.Context) (service.MessageBatch, erro
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode OCF datum to JSON: %s", err)
 	}
-	return service.MessageBatch{service.NewMessage(jb)}, nil
+	msg := service.NewMessage(jb)
+	msg.MetaSetMut("avro_schema", c.avroCodec.CanonicalSchema())
+	msg.MetaSetMut("avro_schema_fingerprint", c.avroCodec.Rabin)
+	return service.MessageBatch{msg}, nil
 }
 
 func (c *avroScanner) Close(ctx context.Context) error {