Repository: honeycombio/refinery
Pull request #1586 (2025-06-13 01:22:32)

perf: add lazy-loading MessagePack payload for events
## Which problem is this PR solving?
Refinery is not efficient! In production profiles we often spend over 50% of CPU time allocating and freeing memory, and a good chunk of the rest deserializing messagepack data.

## Short description of the changes
This is the first (and hopefully largest) change on the way to cutting down the amount of time refinery spends deserializing events. In place of the ubiquitous `map[string]any`, this keeps the original serialized data with the event throughout its lifetime, and interrogates it when necessary to retrieve individual fields. The theory here is that ultimately refinery needs to access only a small fraction of fields, and unpacking the rest into distinct scalars on the heap is unnecessary. This is an approach which has served us well with other services.

Added here are the initial container types needed to make this happen, and all the boilerplate interface changes that go with that. Unfortunately I had to bring in a new message pack library, because our old one doesn't know how to work with flat `[]byte` slices efficiently, but it's also not a drop-in replacement. (The old library also hasn't had a commit in two years and is looking like it may be abandoned.)

But, this PR itself does not improve performance. To see initial gains, we'll need to optimize how we do messagepack deserialization, and pre-request some of our fields instead of demanding them all one-by-one with Get(). And this PR is big enough as it is. Longer term we'll also want to leverage this pre-serialized data during re-transmission to peers, and handle otel data efficiently in some way (probably direct transcoding). For the record I think it would be better if we were using otel for refinery's outbound transmission, but that's too big a change if we want to get this done as well.

This is only one behavioral change here, which is that the event size calculation now uses the serialized data when available, which will definitely produce somewhat different numbers than what we do today. Since this didn't break any tests I'm assuming that ok.




Patch:
diff --git a/collect/collect.go b/collect/collect.go
index 0145e9f..f106294 100644
--- a/collect/collect.go
+++ b/collect/collect.go
@@ -549,13 +549,10 @@ func (i *InMemCollector) redistributeTraces(ctx context.Context) {
 
 			sp.APIHost = newTarget.GetAddress()
 
-			if sp.Data == nil {
-				sp.Data = make(map[string]interface{})
-			}
-			if v, ok := sp.Data["meta.refinery.forwarded"]; ok {
-				sp.Data["meta.refinery.forwarded"] = fmt.Sprintf("%s,%s", v, i.hostname)
+			if v := sp.Data.Get("meta.refinery.forwarded"); v != nil {
+				sp.Data.Set("meta.refinery.forwarded", fmt.Sprintf("%s,%s", v, i.hostname))
 			} else {
-				sp.Data["meta.refinery.forwarded"] = i.hostname
+				sp.Data.Set("meta.refinery.forwarded", i.hostname)
 			}
 
 			i.PeerTransmission.EnqueueSpan(sp)
@@ -673,7 +670,7 @@ func (i *InMemCollector) sendExpiredTracesInCache(ctx context.Context, now time.
 				Dataset: trace.Dataset,
 			},
 		}, trace, i.Sharder.WhichShard(trace.ID()))
-		dc.Data["meta.refinery.expired_trace"] = true
+		dc.Data.Set("meta.refinery.expired_trace", true)
 		i.PeerTransmission.EnqueueEvent(dc)
 	}
 	span.SetAttributes(attribute.Int64("total_spans_sent", totalSpansSent))
@@ -727,7 +724,7 @@ func (i *InMemCollector) processSpan(ctx context.Context, sp *types.Span, source
 
 		// if the span is sent for signaling expired traces,
 		// we should not add it to the cache
-		if sp.Data["meta.refinery.expired_trace"] != nil {
+		if sp.Data.Exists("meta.refinery.expired_trace") {
 			return
 		}
 
@@ -777,7 +774,7 @@ func (i *InMemCollector) processSpan(ctx context.Context, sp *types.Span, source
 
 	// if the span is sent for signaling expired traces,
 	// we should not add it to the cache
-	if sp.Data["meta.refinery.expired_trace"] != nil {
+	if sp.Data.Exists("meta.refinery.expired_trace") {
 		return
 	}
 
@@ -884,12 +881,12 @@ func (i *InMemCollector) ProcessSpanImmediately(sp *types.Span) (processed bool,
 
 	i.Metrics.Increment("kept_from_stress")
 	// ok, we're sending it, so decorate it first
-	sp.Data["meta.stressed"] = true
+	sp.Data.Set("meta.stressed", true)
 	if i.Config.GetAddRuleReasonToTrace() {
-		sp.Data["meta.refinery.reason"] = reason
+		sp.Data.Set("meta.refinery.reason", reason)
 	}
 	if i.hostname != "" {
-		sp.Data["meta.refinery.local_hostname"] = i.hostname
+		sp.Data.Set("meta.refinery.local_hostname", i.hostname)
 	}
 
 	i.addAdditionalAttributes(sp)
@@ -935,12 +932,12 @@ func (i *InMemCollector) dealWithSentTrace(ctx context.Context, tr cache.TraceSe
 		} else {
 			metaReason = "late arriving span"
 		}
-		sp.Data["meta.refinery.reason"] = metaReason
-		sp.Data["meta.refinery.send_reason"] = TraceSendLateSpan
+		sp.Data.Set("meta.refinery.reason", metaReason)
+		sp.Data.Set("meta.refinery.send_reason", TraceSendLateSpan)
 
 	}
 	if i.hostname != "" {
-		sp.Data["meta.refinery.local_hostname"] = i.hostname
+		sp.Data.Set("meta.refinery.local_hostname", i.hostname)
 	}
 	isDryRun := i.Config.GetIsDryRun()
 	keep := tr.Kept()
@@ -951,7 +948,7 @@ func (i *InMemCollector) dealWithSentTrace(ctx context.Context, tr cache.TraceSe
 
 	if isDryRun {
 		// if dry run mode is enabled, we keep all traces and mark the spans with the sampling decision
-		sp.Data[config.DryRunFieldName] = keep
+		sp.Data.Set(config.DryRunFieldName, keep)
 		if !keep {
 			i.Logger.Debug().WithField("trace_id", sp.TraceID).Logf("Sending span that would have been dropped, but dry run mode is enabled")
 			i.Metrics.Increment(TraceSendLateSpan)
@@ -966,12 +963,12 @@ func (i *InMemCollector) dealWithSentTrace(ctx context.Context, tr cache.TraceSe
 		// if this span is a late root span, possibly update it with our current span count
 		if sp.IsRoot {
 			if i.Config.GetAddCountsToRoot() {
-				sp.Data["meta.span_event_count"] = int64(tr.SpanEventCount())
-				sp.Data["meta.span_link_count"] = int64(tr.SpanLinkCount())
-				sp.Data["meta.span_count"] = int64(tr.SpanCount())
-				sp.Data["meta.event_count"] = int64(tr.DescendantCount())
+				sp.Data.Set("meta.span_event_count", int64(tr.SpanEventCount()))
+				sp.Data.Set("meta.span_link_count", int64(tr.SpanLinkCount()))
+				sp.Data.Set("meta.span_count", int64(tr.SpanCount()))
+				sp.Data.Set("meta.event_count", int64(tr.DescendantCount()))
 			} else if i.Config.GetAddSpanCountToRoot() {
-				sp.Data["meta.span_count"] = int64(tr.DescendantCount())
+				sp.Data.Set("meta.span_count", int64(tr.DescendantCount()))
 			}
 		}
 		otelutil.AddSpanField(span, "is_root_span", sp.IsRoot)
@@ -988,7 +985,7 @@ func mergeTraceAndSpanSampleRates(sp *types.Span, traceSampleRate uint, dryRunMo
 	if sp.SampleRate != 0 {
 		// Write down the original sample rate so that that information
 		// is more easily recovered
-		sp.Data["meta.refinery.original_sample_rate"] = sp.SampleRate
+		sp.Data.Set("meta.refinery.original_sample_rate", sp.SampleRate)
 	}
 
 	if tempSampleRate < 1 {
@@ -1003,7 +1000,7 @@ func mergeTraceAndSpanSampleRates(sp *types.Span, traceSampleRate uint, dryRunMo
 	// if spans are already sampled, take that into account when computing
 	// the final rate
 	if dryRunMode {
-		sp.Data["meta.dryrun.sample_rate"] = tempSampleRate * traceSampleRate
+		sp.Data.Set("meta.dryrun.sample_rate", tempSampleRate*traceSampleRate)
 		sp.SampleRate = tempSampleRate
 	} else {
 		sp.SampleRate = tempSampleRate * traceSampleRate
@@ -1041,12 +1038,12 @@ func (i *InMemCollector) send(ctx context.Context, trace *types.Trace, td *Trace
 		rs := trace.RootSpan
 		if rs != nil {
 			if i.Config.GetAddCountsToRoot() {
-				rs.Data["meta.span_event_count"] = int64(td.EventCount)
-				rs.Data["meta.span_link_count"] = int64(td.LinkCount)
-				rs.Data["meta.span_count"] = int64(td.Count)
-				rs.Data["meta.event_count"] = int64(td.DescendantCount())
+				rs.Data.Set("meta.span_event_count", int64(td.EventCount))
+				rs.Data.Set("meta.span_link_count", int64(td.LinkCount))
+				rs.Data.Set("meta.span_count", int64(td.Count))
+				rs.Data.Set("meta.event_count", int64(td.DescendantCount()))
 			} else if i.Config.GetAddSpanCountToRoot() {
-				rs.Data["meta.span_count"] = int64(td.DescendantCount())
+				rs.Data.Set("meta.span_count", int64(td.DescendantCount()))
 			}
 		}
 	}
@@ -1217,9 +1214,6 @@ func (i *InMemCollector) distributeSpansOnShutdown(sentSpanChan chan sentRecord,
 			}
 
 			if sendBy != nil {
-				if sp.Data == nil {
-					sp.Data = make(map[string]interface{})
-				}
 				sp.SetSendBy(*sendBy)
 			}
 			// if there's no trace decision, then we need to forward the trace to its new home
@@ -1246,7 +1240,7 @@ func (i *InMemCollector) sendSpansOnShutdown(ctx context.Context, sentSpanChan <
 			}
 
 			ctx, span := otelutil.StartSpanMulti(ctx, i.Tracer, "shutdown_sent_span", map[string]interface{}{"trace_id": r.span.TraceID, "hostname": i.hostname})
-			r.span.Data["meta.refinery.shutdown.send"] = true
+			r.span.Data.Set("meta.refinery.shutdown.send", true)
 
 			i.dealWithSentTrace(ctx, r.record, r.reason, r.span)
 			_, exist := sentTraces[r.span.TraceID]
@@ -1272,13 +1266,10 @@ func (i *InMemCollector) sendSpansOnShutdown(ctx context.Context, sentSpanChan <
 
 			sp.APIHost = url
 
-			if sp.Data == nil {
-				sp.Data = make(map[string]interface{})
-			}
-			if v, ok := sp.Data["meta.refinery.forwarded"]; ok {
-				sp.Data["meta.refinery.forwarded"] = fmt.Sprintf("%s,%s", v, i.hostname)
+			if v := sp.Data.Get("meta.refinery.forwarded"); v != nil {
+				sp.Data.Set("meta.refinery.forwarded", fmt.Sprintf("%s,%s", v, i.hostname))
 			} else {
-				sp.Data["meta.refinery.forwarded"] = i.hostname
+				sp.Data.Set("meta.refinery.forwarded", i.hostname)
 			}
 
 			i.PeerTransmission.EnqueueSpan(sp)
@@ -1304,7 +1295,7 @@ func (i *InMemCollector) getFromCache(traceID string) *types.Trace {
 
 func (i *InMemCollector) addAdditionalAttributes(sp *types.Span) {
 	for k, v := range i.Config.GetAdditionalAttributes() {
-		sp.Data[k] = v
+		sp.Data.Set(k, v)
 	}
 }
 
@@ -1323,9 +1314,12 @@ func (i *InMemCollector) createDecisionSpan(sp *types.Span, trace *types.Trace,
 	dc := sp.ExtractDecisionContext()
 	// extract all key fields from the span
 	keyFields := sampler.GetKeyFields()
+	sp.Data.MemoizeFields(keyFields...)
 	for _, keyField := range keyFields {
-		if val, ok := sp.Data[keyField]; ok {
-			dc.Data[keyField] = val
+		// Less efficient than a two-return version of Get(), so consider adding
+		// that to the Payload interface if these becomes a hotspot.
+		if sp.Data.Exists(keyField) {
+			dc.Data.Set(keyField, sp.Data.Get(keyField))
 		}
 	}
 
@@ -1356,10 +1350,10 @@ func (i *InMemCollector) sendTraces() {
 			}
 
 			if i.Config.GetAddRuleReasonToTrace() {
-				sp.Data["meta.refinery.reason"] = t.reason
-				sp.Data["meta.refinery.send_reason"] = t.sendReason
+				sp.Data.Set("meta.refinery.reason", t.reason)
+				sp.Data.Set("meta.refinery.send_reason", t.sendReason)
 				if t.sampleKey != "" {
-					sp.Data["meta.refinery.sample_key"] = t.sampleKey
+					sp.Data.Set("meta.refinery.sample_key", t.sampleKey)
 				}
 			}
 
@@ -1367,21 +1361,21 @@ func (i *InMemCollector) sendTraces() {
 			// with the final total as of our send time
 			if sp.IsRoot {
 				if i.Config.GetAddCountsToRoot() {
-					sp.Data["meta.span_event_count"] = int64(t.SpanEventCount())
-					sp.Data["meta.span_link_count"] = int64(t.SpanLinkCount())
-					sp.Data["meta.span_count"] = int64(t.SpanCount())
-					sp.Data["meta.event_count"] = int64(t.DescendantCount())
+					sp.Data.Set("meta.span_event_count", int64(t.SpanEventCount()))
+					sp.Data.Set("meta.span_link_count", int64(t.SpanLinkCount()))
+					sp.Data.Set("meta.span_count", int64(t.SpanCount()))
+					sp.Data.Set("meta.event_count", int64(t.DescendantCount()))
 				} else if i.Config.GetAddSpanCountToRoot() {
-					sp.Data["meta.span_count"] = int64(t.DescendantCount())
+					sp.Data.Set("meta.span_count", int64(t.DescendantCount()))
 				}
 			}
 
 			isDryRun := i.Config.GetIsDryRun()
 			if isDryRun {
-				sp.Data[config.DryRunFieldName] = t.shouldSend
+				sp.Data.Set(config.DryRunFieldName, t.shouldSend)
 			}
 			if i.hostname != "" {
-				sp.Data["meta.refinery.local_hostname"] = i.hostname
+				sp.Data.Set("meta.refinery.local_hostname", i.hostname)
 			}
 			mergeTraceAndSpanSampleRates(sp, t.SampleRate(), isDryRun)
 			i.addAdditionalAttributes(sp)
diff --git a/go.mod b/go.mod
index 3a455ab..0db1059 100644
--- a/go.mod
+++ b/go.mod
@@ -32,6 +32,7 @@ require (
 	github.com/sourcegraph/conc v0.3.0
 	github.com/stretchr/testify v1.10.0
 	github.com/tidwall/gjson v1.18.0
+	github.com/tinylib/msgp v1.3.0
 	github.com/vmihailenco/msgpack/v5 v5.4.1
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0
@@ -64,6 +65,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.127.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/sampling v0.125.0 // indirect
 	github.com/otiai10/copy v1.10.0 // indirect
+	github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect
 	github.com/sergi/go-diff v1.2.0 // indirect
 	github.com/spf13/cobra v1.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
diff --git a/go.sum b/go.sum
index e4884f7..ec435ab 100644
--- a/go.sum
+++ b/go.sum
@@ -173,6 +173,8 @@ github.com/panmari/cuckoofilter v1.0.6 h1:WKb1aSj16h22x0CKVtTCaRkJiCnVGPLEMGbNY8
 github.com/panmari/cuckoofilter v1.0.6/go.mod h1:bKADbQPGbN6TxUvo/IbMEIUbKuASnpsOvrLTgpSX0aU=
 github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
+github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c h1:dAMKvw0MlJT1GshSTtih8C2gDs04w8dReiOGXrGLNoY=
+github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -227,6 +229,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tinylib/msgp v1.3.0 h1:ULuf7GPooDaIlbyvgAxBV/FI7ynli6LZ1/nVUNu+0ww=
+github.com/tinylib/msgp v1.3.0/go.mod h1:ykjzy2wzgrlvpDCRc4LA8UXy6D8bzMSuAF3WD57Gok0=
 github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
diff --git a/route/route.go b/route/route.go
index 09c11cb..c933fdd 100644
--- a/route/route.go
+++ b/route/route.go
@@ -452,7 +452,7 @@ func (r *Router) requestToEvent(ctx context.Context, req *http.Request, reqBod [
 		Environment: environment,
 		SampleRate:  uint(sampleRate),
 		Timestamp:   eventTime,
-		Data:        data,
+		Data:        types.NewPayload(data),
 	}, nil
 }
 
@@ -569,7 +569,7 @@ func (router *Router) processOTLPRequest(
 				Environment: environment,
 				SampleRate:  uint(ev.SampleRate),
 				Timestamp:   ev.Timestamp,
-				Data:        ev.Attributes,
+				Data:        types.NewPayload(ev.Attributes),
 			}
 			addIncomingUserAgent(event, incomingUserAgent)
 			if err = router.processEvent(event, requestID); err != nil {
@@ -594,7 +594,7 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 	r.Metrics.Histogram(r.incomingOrPeer+"_router_event_bytes", float64(ev.GetDataSize()))
 
 	// check if this is a probe from another refinery; if so, we should drop it
-	if ev.Data["meta.refinery.probe"] != nil {
+	if ev.Data.Get("meta.refinery.probe") != nil {
 		debugLog.Logf("dropping probe")
 		return nil
 	}
@@ -619,7 +619,7 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 
 	// only record bytes received for incoming traffic when opamp is enabled and record usage is set to true
 	if r.incomingOrPeer == "incoming" && r.Config.GetOpAMPConfig().Enabled && r.Config.GetOpAMPConfig().RecordUsage.Get() {
-		if span.Data["meta.signal_type"] == "log" {
+		if span.Data.Get("meta.signal_type") == "log" {
 			r.Metrics.Count("bytes_received_logs", span.GetDataSize())
 		} else {
 			r.Metrics.Count("bytes_received_traces", span.GetDataSize())
@@ -643,7 +643,7 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 
 				// If the span was kept, we want to generate a probe that we'll forward
 				// to a peer IF this span would have been forwarded.
-				ev.Data["meta.refinery.probe"] = true
+				ev.Data.Set("meta.refinery.probe", true)
 				isProbe = true
 			}
 		}
@@ -733,10 +733,10 @@ func (r *Router) getMaybeCompressedBody(req *http.Request) (io.Reader, error) {
 }
 
 type batchedEvent struct {
-	Timestamp        string                 `json:"time"`
-	MsgPackTimestamp *time.Time             `msgpack:"time,omitempty"`
-	SampleRate       int64                  `json:"samplerate" msgpack:"samplerate"`
-	Data             map[string]interface{} `json:"data" msgpack:"data"`
+	Timestamp        string        `json:"time"`
+	MsgPackTimestamp *time.Time    `msgpack:"time,omitempty"`
+	SampleRate       int64         `json:"samplerate" msgpack:"samplerate"`
+	Data             types.Payload `json:"data" msgpack:"data"`
 }
 
 func (b *batchedEvent) getEventTime() time.Time {
@@ -1065,24 +1065,18 @@ func getDatasetFromRequest(req *http.Request) (string, error) {
 
 func isRootSpan(ev *types.Event, cfg config.Config) bool {
 	// log event should never be considered a root span, check for that first
-	if signalType := ev.Data["meta.signal_type"]; signalType == "log" {
+	if signalType := ev.Data.Get("meta.signal_type"); signalType == "log" {
 		return false
 	}
 
 	// check if the event has a root flag
-	if isRoot, ok := ev.Data["meta.refinery.root"]; ok {
-		v, ok := isRoot.(bool)
-		if !ok {
-			return false
-		}
-
-		return v
+	if isRoot, ok := ev.Data.Get("meta.refinery.root").(bool); ok {
+		return isRoot
 	}
 
 	// check if the event has a parent id using the configured parent id field names
 	for _, parentIdFieldName := range cfg.GetParentIdFieldNames() {
-		parentId := ev.Data[parentIdFieldName]
-		if _, ok := parentId.(string); ok && parentId != "" {
+		if parentId, ok := ev.Data.Get(parentIdFieldName).(string); ok && parentId != "" {
 			return false
 		}
 	}
@@ -1090,13 +1084,13 @@ func isRootSpan(ev *types.Event, cfg config.Config) bool {
 }
 
 func extractTraceID(traceIdFieldNames []string, ev *types.Event) string {
-	if trID, ok := ev.Data["meta.trace_id"]; ok {
-		return trID.(string)
+	if trID, ok := ev.Data.Get("meta.trace_id").(string); ok {
+		return trID
 	}
 
 	for _, traceIdFieldName := range traceIdFieldNames {
-		if trID, ok := ev.Data[traceIdFieldName]; ok {
-			return trID.(string)
+		if trID, ok := ev.Data.Get(traceIdFieldName).(string); ok {
+			return trID
 		}
 	}
 
@@ -1108,7 +1102,7 @@ func getUserAgentFromRequest(req *http.Request) string {
 }
 
 func addIncomingUserAgent(ev *types.Event, userAgent string) {
-	if userAgent != "" && ev.Data["meta.refinery.incoming_user_agent"] == nil {
-		ev.Data["meta.refinery.incoming_user_agent"] = userAgent
+	if userAgent != "" && ev.Data.Get("meta.refinery.incoming_user_agent") == nil {
+		ev.Data.Set("meta.refinery.incoming_user_agent", userAgent)
 	}
 }
diff --git a/sample/rules.go b/sample/rules.go
index 14d5b83..d0d7439 100644
--- a/sample/rules.go
+++ b/sample/rules.go
@@ -281,7 +281,8 @@ func extractValueFromSpan(
 	trace *types.Trace,
 	span *types.Span,
 	condition *config.RulesBasedSamplerCondition,
-	checkNestedFields bool) (value interface{}, exists bool, checkedOnlyRoot bool) {
+	checkNestedFields bool,
+) (value interface{}, exists bool, checkedOnlyRoot bool) {
 	// start with the assumption that we only checked the root span
 	checkedOnlyRoot = true
 
@@ -317,12 +318,11 @@ func extractValueFromSpan(
 			checkedOnlyRoot = false
 		}
 
-		value, exists = span.Data[field]
-		if exists {
-			return value, exists, checkedOnlyRoot
+		if span.Data.Exists(field) {
+			return span.Data.Get(field), true, checkedOnlyRoot
 		}
 	}
-	if !exists && checkNestedFields {
+	if checkNestedFields {
 		jsonStr, err := json.Marshal(span.Data)
 		if err == nil {
 			for _, field := range condition.Fields {
diff --git a/sample/trace_key.go b/sample/trace_key.go
index f10a290..ab6d056 100644
--- a/sample/trace_key.go
+++ b/sample/trace_key.go
@@ -60,11 +60,11 @@ func (d *traceKey) build(trace *types.Trace) (string, int) {
 outer:
 	for i, field := range d.fields {
 		for _, span := range spans {
-			if val, ok := span.Data[field]; ok {
+			if span.Data.Exists(field) {
 				if d.distinctValue.totalUniqueCount >= maxKeyLength {
 					break outer
 				}
-				d.distinctValue.AddAsString(val, i)
+				d.distinctValue.AddAsString(span.Data.Get(field), i)
 			}
 		}
 	}
@@ -92,8 +92,8 @@ outer:
 
 	if trace.RootSpan != nil {
 		for _, field := range d.rootOnlyFields {
-			if val, ok := trace.RootSpan.Data[field]; ok {
-				d.keyBuilder.WriteString(fmt.Sprintf("%v,", val))
+			if trace.RootSpan.Data.Exists(field) {
+				d.keyBuilder.WriteString(fmt.Sprintf("%v,", trace.RootSpan.Data.Get(field)))
 				fieldCount += 1
 			}
 		}
diff --git a/transmit/transmit.go b/transmit/transmit.go
index 3fdfe9a..de19e28 100644
--- a/transmit/transmit.go
+++ b/transmit/transmit.go
@@ -97,7 +97,7 @@ func (d *DefaultTransmission) EnqueueEvent(ev *types.Event) {
 		WithString("api_host", ev.APIHost).
 		WithString("dataset", ev.Dataset).
 		Logf("transmit sending event")
-	libhEv := d.builder.NewEventSized(len(ev.Data))
+	libhEv := d.builder.NewEventSized(ev.Data.GetDataSize())
 	libhEv.APIHost = ev.APIHost
 	libhEv.WriteKey = ev.APIKey
 	libhEv.Dataset = ev.Dataset
@@ -112,13 +112,13 @@ func (d *DefaultTransmission) EnqueueEvent(ev *types.Event) {
 	}
 
 	for _, k := range d.Config.GetAdditionalErrorFields() {
-		if v, ok := ev.Data[k]; ok {
-			metadata[k] = v
+		if ev.Data.Exists(k) {
+			metadata[k] = ev.Data.Get(k)
 		}
 	}
 	libhEv.Metadata = metadata
 
-	for k, v := range ev.Data {
+	for k, v := range ev.Data.All() {
 		libhEv.AddField(k, v)
 	}
 
diff --git a/types/event.go b/types/event.go
index 5322186..4f47cfd 100644
--- a/types/event.go
+++ b/types/event.go
@@ -30,50 +30,18 @@ type Event struct {
 	Environment string
 	SampleRate  uint
 	Timestamp   time.Time
-	Data        map[string]interface{}
+	Data        Payload
 	dataSize    int
 }
 
 // GetDataSize computes the size of the Data element of the Event.
 func (e *Event) GetDataSize() int {
 	if e.dataSize == 0 {
-		for k, v := range e.Data {
-			e.dataSize += len(k) + getByteSize(v)
-		}
+		e.dataSize = e.Data.GetDataSize()
 	}
 	return e.dataSize
 }
 
-// getByteSize returns the size of the given value in bytes.
-// This is a rough estimate, but it's good enough for our purposes.
-// Maps and slices are the most complex, so we'll just add up the sizes of their entries.
-func getByteSize(val any) int {
-	switch value := val.(type) {
-	case bool:
-		return 1
-	case float64, int64, int:
-		return 8
-	case string:
-		return len(value)
-	case []byte: // also catch []uint8
-		return len(value)
-	case []any:
-		total := 0
-		for _, v := range value {
-			total += getByteSize(v)
-		}
-		return total
-	case map[string]any:
-		total := 0
-		for k, v := range value {
-			total += len(k) + getByteSize(v)
-		}
-		return total
-	default:
-		return 8 // catchall
-	}
-}
-
 // Trace isn't something that shows up on the wire; it gets created within
 // Refinery. Traces are not thread-safe; only one goroutine should be working
 // with a trace object at a time.
@@ -267,14 +235,7 @@ type Span struct {
 // IsDecicionSpan returns true if the span is a decision span based on
 // a flag set in the span's metadata.
 func (sp *Span) IsDecisionSpan() bool {
-	if sp.Data == nil {
-		return false
-	}
-	v, ok := sp.Data["meta.refinery.min_span"]
-	if !ok {
-		return false
-	}
-	isDecisionSpan, ok := v.(bool)
+	isDecisionSpan, ok := sp.Data.Get("meta.refinery.min_span").(bool)
 	if !ok {
 		return false
 	}
@@ -287,7 +248,7 @@ func (sp *Span) IsDecisionSpan() bool {
 func (sp *Span) ExtractDecisionContext() *Event {
 	decisionCtx := sp.Event
 	dataSize := sp.Event.GetDataSize()
-	decisionCtx.Data = map[string]interface{}{
+	decisionData := map[string]interface{}{
 		"meta.trace_id":                sp.TraceID,
 		"meta.refinery.root":           sp.IsRoot,
 		"meta.refinery.min_span":       true,
@@ -296,27 +257,23 @@ func (sp *Span) ExtractDecisionContext() *Event {
 	}
 
 	if v, ok := sp.GetSendBy(); ok {
-		decisionCtx.Data["meta.refinery.send_by"] = v
+		decisionData["meta.refinery.send_by"] = v
 	}
+	decisionCtx.Data = NewPayload(decisionData)
 	return &decisionCtx
 }
 
 func (sp *Span) SetSendBy(sendBy time.Time) {
-	sp.Data["meta.refinery.send_by"] = sendBy.Unix()
+	sp.Data.Set("meta.refinery.send_by", sendBy.Unix())
 }
 
 func (sp *Span) GetSendBy() (time.Time, bool) {
-	if sp.Data == nil {
-		return time.Time{}, false
-	}
-
-	if value, ok := sp.Data["meta.refinery.send_by"]; ok {
-		switch v := value.(type) {
-		case int64:
-			return time.Unix(v, 0), true
-		case uint64:
-			return time.Unix(int64(v), 0), true
-		}
+	value := sp.Data.Get("meta.refinery.send_by")
+	switch v := value.(type) {
+	case int64:
+		return time.Unix(v, 0), true
+	case uint64:
+		return time.Unix(int64(v), 0), true
 	}
 
 	return time.Time{}, false
@@ -327,13 +284,12 @@ func (sp *Span) GetSendBy() (time.Time, bool) {
 // relative ordering, not absolute calculations.
 func (sp *Span) GetDataSize() int {
 	if sp.IsDecisionSpan() {
-		if v, ok := sp.Data["meta.refinery.span_data_size"]; ok {
-			switch value := v.(type) {
-			case int64:
-				return int(value)
-			case uint64:
-				return int(value)
-			}
+		v := sp.Data.Get("meta.refinery.span_data_size")
+		switch value := v.(type) {
+		case int64:
+			return int(value)
+		case uint64:
+			return int(value)
 		}
 		return 0
 	}
@@ -360,7 +316,7 @@ func (sp *Span) AnnotationType() SpanAnnotationType {
 	if sp.annotationType != SpanAnnotationTypeUnSet {
 		return sp.annotationType
 	}
-	t := sp.Data["meta.annotation_type"]
+	t := sp.Data.Get("meta.annotation_type")
 	switch t {
 	case "span_event":
 		sp.annotationType = SpanAnnotationTypeSpanEvent
diff --git a/types/msgp_payload_map.go b/types/msgp_payload_map.go
new file mode 100644
index 0000000..a62e4d8
--- /dev/null
+++ b/types/msgp_payload_map.go
@@ -0,0 +1,192 @@
+package types
+
+import (
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/tinylib/msgp/msgp"
+)
+
+type FieldType int
+
+const (
+	FieldTypeUnknown = iota
+	FieldTypeInt64
+	FieldTypeFloat64
+	FieldTypeString
+	FieldTypeBool
+
+	// Arrays, maps, other stuff supported by the wire protocols but not
+	// expected to be very common.
+	FieldTypeOther
+)
+
+// A wrapper for a serialized messagepack map. Reading its values can be far
+// cheaper than deserializing into a real map, but the gains disappear quickly
+// if you do too many iterations. It is recommended to retrieve values in batches
+// and memoize anything that will be used more than once.
+type MsgpPayloadMap struct {
+	rawData []byte
+}
+
+func NewMessagePackPayloadMap(raw []byte) MsgpPayloadMap {
+	return MsgpPayloadMap{
+		rawData: raw,
+	}
+}
+
+func (m *MsgpPayloadMap) Size() int {
+	return len(m.rawData)
+}
+
+func (m *MsgpPayloadMap) Iterate() (msgpPayloadMapIter, error) {
+	if m == nil {
+		return msgpPayloadMapIter{}, nil
+	}
+	n, remaining, err := msgp.ReadMapHeaderBytes(m.rawData)
+	if err != nil {
+		return msgpPayloadMapIter{}, err
+	}
+	if n < 1 {
+		return msgpPayloadMapIter{}, nil
+	}
+
+	return msgpPayloadMapIter{
+		remaining: remaining,
+	}, nil
+}
+
+type msgpPayloadMapIter struct {
+	remaining    []byte
+	pendingValue bool
+}
+
+// Returns the key string as []byte and type of the next map value.
+// Note that the key slice may be invalidated by the next call to NextKey()
+// or completing iteration. It must not be stored.
+// Returns EOF when finished.
+func (m *msgpPayloadMapIter) NextKey() (key []byte, typ FieldType, err error) {
+	// The last value was never consumed, that's common. Skip it.
+	if m.pendingValue {
+		m.remaining, err = msgp.Skip(m.remaining)
+		if err != nil {
+			return nil, FieldTypeUnknown, err
+		}
+	}
+
+	if len(m.remaining) == 0 {
+		return nil, FieldTypeUnknown, io.EOF
+	}
+
+	key, m.remaining, err = msgp.ReadMapKeyZC(m.remaining)
+	if err != nil {
+		return nil, FieldTypeUnknown, err
+	}
+
+	typ, err = msgpTypeToFieldType(msgp.NextType(m.remaining))
+	if err != nil {
+		return nil, FieldTypeUnknown, err
+	}
+
+	m.pendingValue = true
+	return key, typ, nil
+}
+
+// The Value functions return the value corresponding to the previous call to
+// NextKey(), if any. ValueAny() returns any type as any.
+func (m *msgpPayloadMapIter) ValueAny() (any, error) {
+	if !m.pendingValue {
+		return nil, errors.New("no pending value")
+	}
+	m.pendingValue = false
+	val, remaining, err := msgp.ReadIntfBytes(m.remaining)
+	if err == nil {
+		m.remaining = remaining
+	} else {
+		m.remaining, _ = msgp.Skip(m.remaining)
+	}
+	return val, err
+}
+
+// The typed values decode data as the requested type, if possible, but don't
+// attempt to coerce non-matching types. If the value is not of the expected
+// type, returns an error.
+func (m *msgpPayloadMapIter) ValueInt64() (int64, error) {
+	if !m.pendingValue {
+		return 0, errors.New("no pending value")
+	}
+	m.pendingValue = false
+	val, remaining, err := msgp.ReadInt64Bytes(m.remaining)
+	if err == nil {
+		m.remaining = remaining
+	} else {
+		m.remaining, _ = msgp.Skip(m.remaining)
+	}
+	return val, err
+}
+
+func (m *msgpPayloadMapIter) ValueFloat64() (float64, error) {
+	if !m.pendingValue {
+		return 0, errors.New("no pending value")
+	}
+	m.pendingValue = false
+	val, remaining, err := msgp.ReadFloat64Bytes(m.remaining)
+	if err == nil {
+		m.remaining = remaining
+	} else {
+		m.remaining, _ = msgp.Skip(m.remaining)
+	}
+	return val, err
+}
+
+func (m *msgpPayloadMapIter) ValueBool() (bool, error) {
+	if !m.pendingValue {
+		return false, errors.New("no pending value")
+	}
+	m.pendingValue = false
+	val, remaining, err := msgp.ReadBoolBytes(m.remaining)
+	if err == nil {
+		m.remaining = remaining
+	} else {
+		m.remaining, _ = msgp.Skip(m.remaining)
+	}
+	return val, err
+}
+
+func (m *msgpPayloadMapIter) ValueString() (string, error) {
+	if !m.pendingValue {
+		return "", errors.New("no pending value")
+	}
+	m.pendingValue = false
+	val, remaining, err := msgp.ReadStringBytes(m.remaining)
+	if err == nil {
+		m.remaining = remaining
+	} else {
+		m.remaining, _ = msgp.Skip(m.remaining)
+	}
+	return val, err
+}
+
+func msgpTypeToFieldType(t msgp.Type) (FieldType, error) {
+	switch t {
+	case msgp.StrType, msgp.BinType:
+		return FieldTypeString, nil
+
+	case msgp.MapType, msgp.ArrayType, msgp.NilType, msgp.DurationType,
+		msgp.ExtensionType, msgp.Complex64Type, msgp.Complex128Type, msgp.TimeType:
+		return FieldTypeOther, nil
+
+	case msgp.Float64Type, msgp.Float32Type, msgp.NumberType:
+		return FieldTypeFloat64, nil
+
+	case msgp.BoolType:
+		return FieldTypeBool, nil
+
+	case msgp.IntType, msgp.UintType:
+		return FieldTypeInt64, nil
+
+	default:
+		return FieldTypeUnknown, fmt.Errorf("msgpack: unknown msgp type %x", t)
+	}
+}
diff --git a/types/payload.go b/types/payload.go
new file mode 100644
index 0000000..a8bee6f
--- /dev/null
+++ b/types/payload.go
@@ -0,0 +1,262 @@
+package types
+
+import (
+	"encoding/json"
+	"iter"
+	"maps"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+type Payload struct {
+	// A serialized messagepack map used to source fields.
+	msgpMap *MsgpPayloadMap
+
+	// Deserialized fields, either from the internal msgpMap, or set externally.
+	memoizedFields map[string]any
+}
+
+func NewPayload(data map[string]any) Payload {
+	return Payload{
+		memoizedFields: data,
+	}
+}
+
+func (p *Payload) UnmarshalMsgpack(data []byte) error {
+	p.msgpMap = &MsgpPayloadMap{rawData: data}
+	p.memoizedFields = make(map[string]any)
+	return nil
+}
+
+func (p *Payload) UnmarshalJSON(data []byte) error {
+	var fields map[string]any
+	if err := jsoniter.Unmarshal(data, &fields); err != nil {
+		return err
+	}
+	p.memoizedFields = fields
+	return nil
+}
+
+// Extracts all of the listed fields from the internal msgp buffer in a single
+// pass, for efficient random access later.
+func (p *Payload) MemoizeFields(keys ...string) {
+	if p.msgpMap == nil {
+		return
+	}
+
+	if p.memoizedFields == nil {
+		p.memoizedFields = make(map[string]any, len(keys))
+	}
+
+	keysToFind := make(map[string]struct{}, len(keys))
+	for _, key := range keys {
+		if _, ok := p.memoizedFields[key]; !ok {
+			keysToFind[key] = struct{}{}
+		}
+	}
+	if len(keysToFind) == 0 {
+		return
+	}
+
+	iter, err := p.msgpMap.Iterate()
+	if err != nil {
+		return
+	}
+
+	var keysFound int
+	for keysFound < len(keysToFind) {
+		keyBytes, _, err := iter.NextKey()
+		if err != nil {
+			break
+		}
+
+		// Note we deliberately don't put string(keyBytes) in a variable here,
+		// because doing so will move it to the heap on every iteration.
+		// Keeping the string cast inline like this allows us to avoid the heap
+		// unless we're actually going to memoize the field.
+		if _, ok := keysToFind[string(keyBytes)]; ok {
+			value, err := iter.ValueAny()
+			if err != nil {
+				break
+			}
+			key := string(keyBytes)
+			p.memoizedFields[key] = value
+			keysFound++
+		}
+	}
+}
+
+func (p *Payload) Exists(key string) bool {
+	if p.memoizedFields != nil {
+		if _, ok := p.memoizedFields[key]; ok {
+			return true
+		}
+	}
+
+	iter, err := p.msgpMap.Iterate()
+	if err != nil {
+		return false
+	}
+
+	for {
+		keyBytes, _, err := iter.NextKey()
+		if err != nil {
+			break
+		}
+
+		if string(keyBytes) == key {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (p *Payload) Get(key string) any {
+	if p.memoizedFields != nil {
+		if value, ok := p.memoizedFields[key]; ok {
+			return value
+		}
+	}
+
+	iter, err := p.msgpMap.Iterate()
+	if err != nil {
+		return nil
+	}
+
+	for {
+		keyBytes, _, err := iter.NextKey()
+		if err != nil {
+			break
+		}
+
+		if string(keyBytes) == key {
+			value, err := iter.ValueAny()
+			if err == nil {
+				return value
+			}
+			break
+		}
+	}
+
+	return nil
+}
+
+func (p *Payload) Set(key string, value any) {
+	if p.memoizedFields == nil {
+		p.memoizedFields = make(map[string]any)
+	}
+
+	p.memoizedFields[key] = value
+}
+
+func (p *Payload) IsEmpty() bool {
+	return len(p.memoizedFields) == 0 && p.msgpMap == nil
+}
+
+// All() allows easily iterating all values in the Payload, but this is very
+// NOT EFFICIENT relative to getting a subset of values using Get. Don't use
+// this in non-test code unless you have to other choice.
+// We only expect this to happen when transmitting sampled events using Libhoney.
+func (p *Payload) All() iter.Seq2[string, any] {
+	return func(yield func(string, any) bool) {
+		// First yield memoized fields
+		for key, value := range p.memoizedFields {
+			if !yield(key, value) {
+				return
+			}
+		}
+
+		// Then iterate through msgpMap for any remaining fields
+		iter, err := p.msgpMap.Iterate()
+		if err != nil {
+			return
+		}
+
+		for {
+			keyBytes, _, err := iter.NextKey()
+			if err != nil {
+				break
+			}
+
+			key := string(keyBytes)
+			if p.memoizedFields != nil {
+				if _, ok := p.memoizedFields[key]; ok {
+					// Don't yield the same key twice. Memoized values take
+					// precedence over serialized values, which we can't
+					// update.
+					continue
+				}
+			}
+
+			value, err := iter.ValueAny()
+			if err != nil {
+				return
+			}
+			if !yield(key, value) {
+				return
+			}
+		}
+	}
+}
+
+// Estimates data size, not very accurately, but it's fast.
+func (p *Payload) GetDataSize() int {
+	var total int
+	if p.msgpMap != nil {
+		total += p.msgpMap.Size()
+	}
+	for k, v := range p.memoizedFields {
+		total += len(k) + getByteSize(v)
+	}
+	return total
+}
+
+// getByteSize returns the size of the given value in bytes.
+// This is a rough estimate, but it's good enough for our purposes.
+// Maps and slices are the most complex, so we'll just add up the sizes of their entries.
+func getByteSize(val any) int {
+	switch value := val.(type) {
+	case bool:
+		return 1
+	case float64, int64, int:
+		return 8
+	case string:
+		return len(value)
+	case []byte: // also catch []uint8
+		return len(value)
+	case []any:
+		total := 0
+		for _, v := range value {
+			total += getByteSize(v)
+		}
+		return total
+	case map[string]any:
+		total := 0
+		for k, v := range value {
+			total += len(k) + getByteSize(v)
+		}
+		return total
+	default:
+		return 8 // catchall
+	}
+}
+
+// MarshalJSON implements json.Marshaler to serialize the Payload as a single JSON object
+// containing all fields from both memoizedFields and msgpMap. This is incredibly
+// inefficient and is only here (for now) to support our legacy nested field implementation.
+func (p Payload) MarshalJSON() ([]byte, error) {
+	return json.Marshal(maps.Collect(p.All()))
+}
+
+// Inefficient, only here for test cases where we serialize a Payload field.
+func (p Payload) MarshalMsgpack() ([]byte, error) {
+	return msgpack.Marshal(maps.Collect(p.All()))
+}
+
+// For debugging purposes
+func (p Payload) String() string {
+	buf, _ := p.MarshalJSON()
+	return string(buf)
+}


Test patch:
diff --git a/collect/collect_test.go b/collect/collect_test.go
index 2166835..9ec86a7 100644
--- a/collect/collect_test.go
+++ b/collect/collect_test.go
@@ -208,9 +208,9 @@ func TestAddRootSpan(t *testing.T) {
 		Event: types.Event{
 			Dataset: "aoeu",
 			APIKey:  legacyAPIKey,
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"meta.refinery.min_span": true,
-			},
+			}),
 		},
 		IsRoot: true,
 	}
@@ -305,7 +305,7 @@ func TestOriginalSampleRateIsNotedInMetaField(t *testing.T) {
 				Dataset:    "aoeu",
 				APIKey:     legacyAPIKey,
 				SampleRate: originalSampleRate,
-				Data:       make(map[string]interface{}),
+				Data:       types.NewPayload(make(map[string]interface{})),
 			},
 			IsRoot: true,
 		}
@@ -319,7 +319,7 @@ func TestOriginalSampleRateIsNotedInMetaField(t *testing.T) {
 	upstreamSampledEvent := events[0]
 
 	assert.NotNil(t, upstreamSampledEvent)
-	assert.Equal(t, originalSampleRate, upstreamSampledEvent.Data["meta.refinery.original_sample_rate"],
+	assert.Equal(t, originalSampleRate, upstreamSampledEvent.Data.Get("meta.refinery.original_sample_rate"),
 		"metadata should be populated with original sample rate")
 	assert.Equal(t, originalSampleRate*uint(expectedDeterministicSampleRate), upstreamSampledEvent.SampleRate,
 		"sample rate for the event should be the original sample rate multiplied by the deterministic sample rate")
@@ -331,7 +331,7 @@ func TestOriginalSampleRateIsNotedInMetaField(t *testing.T) {
 			Dataset:    "no-upstream-sampling",
 			APIKey:     legacyAPIKey,
 			SampleRate: 0, // no upstream sampling
-			Data:       make(map[string]interface{}),
+			Data:       types.NewPayload(make(map[string]interface{})),
 		},
 		IsRoot: true,
 	})
@@ -351,7 +351,7 @@ func TestOriginalSampleRateIsNotedInMetaField(t *testing.T) {
 
 	require.NotNil(t, noUpstreamSampleRateEvent)
 	assert.Equal(t, "no-upstream-sampling", noUpstreamSampleRateEvent.Dataset)
-	assert.Nil(t, noUpstreamSampleRateEvent.Data["meta.refinery.original_sample_rate"],
+	assert.Nil(t, noUpstreamSampleRateEvent.Data.Get("meta.refinery.original_sample_rate"),
 		"original sample rate should not be set in metadata when original sample rate is zero")
 }
 
@@ -404,7 +404,7 @@ func TestTransmittedSpansShouldHaveASampleRateOfAtLeastOne(t *testing.T) {
 			Dataset:    "aoeu",
 			APIKey:     legacyAPIKey,
 			SampleRate: 0, // This should get lifted to 1
-			Data:       make(map[string]interface{}),
+			Data:       types.NewPayload(make(map[string]interface{})),
 		},
 		IsRoot: true,
 	}
@@ -462,9 +462,9 @@ func TestAddSpan(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -482,11 +482,11 @@ func TestAddSpan(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id":        "unused",
 				"meta.refinery.min_span": true,
 				"meta.refinery.send_by":  sendBy,
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -505,7 +505,7 @@ func TestAddSpan(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -585,7 +585,7 @@ func TestDryRunMode(t *testing.T) {
 	span := &types.Span{
 		TraceID: traceID1,
 		Event: types.Event{
-			Data:   map[string]interface{}{},
+			Data:   types.NewPayload(map[string]interface{}{}),
 			APIKey: legacyAPIKey,
 		},
 		IsRoot: true,
@@ -598,7 +598,7 @@ func TestDryRunMode(t *testing.T) {
 	// * remove the trace from the cache
 	events := transmission.GetBlock(1)
 	require.Equal(t, 1, len(events), "adding a root span should send the span")
-	assert.Equal(t, keepTraceID1, events[0].Data[config.DryRunFieldName], "config.DryRunFieldName should match sampling decision for its trace ID")
+	assert.Equal(t, keepTraceID1, events[0].Data.Get(config.DryRunFieldName), "config.DryRunFieldName should match sampling decision for its trace ID")
 	assert.Nil(t, coll.getFromCache(traceID1), "after sending the span, it should be removed from the cache")
 
 	// add a non-root span, create the trace in the cache
@@ -606,9 +606,9 @@ func TestDryRunMode(t *testing.T) {
 		TraceID: traceID2,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -621,7 +621,7 @@ func TestDryRunMode(t *testing.T) {
 	span = &types.Span{
 		TraceID: traceID2,
 		Event: types.Event{
-			Data:   map[string]interface{}{},
+			Data:   types.NewPayload(map[string]interface{}{}),
 			APIKey: legacyAPIKey,
 		},
 		IsRoot: true,
@@ -632,18 +632,18 @@ func TestDryRunMode(t *testing.T) {
 	events = transmission.GetBlock(2)
 	require.Equal(t, 2, len(events), "adding another root span should send the span")
 	// both spanscollectshould be marked with the sampling decision
-	assert.Equal(t, keepTraceID2, events[0].Data[config.DryRunFieldName], "config.DryRunFieldName should match sampling decision for its trace ID")
-	assert.Equal(t, keepTraceID2, events[1].Data[config.DryRunFieldName], "config.DryRunFieldName should match sampling decision for its trace ID")
+	assert.Equal(t, keepTraceID2, events[0].Data.Get(config.DryRunFieldName), "config.DryRunFieldName should match sampling decision for its trace ID")
+	assert.Equal(t, keepTraceID2, events[1].Data.Get(config.DryRunFieldName), "config.DryRunFieldName should match sampling decision for its trace ID")
 	// check that meta value associated with dry run mode is properly applied
-	assert.Equal(t, uint(10), events[0].Data["meta.dryrun.sample_rate"])
+	assert.Equal(t, uint(10), events[0].Data.Get("meta.dryrun.sample_rate"))
 	// check expected sampleRate against span data
-	assert.Equal(t, sampleRate1, events[0].Data["meta.dryrun.sample_rate"])
-	assert.Equal(t, sampleRate2, events[1].Data["meta.dryrun.sample_rate"])
+	assert.Equal(t, sampleRate1, events[0].Data.Get("meta.dryrun.sample_rate"))
+	assert.Equal(t, sampleRate2, events[1].Data.Get("meta.dryrun.sample_rate"))
 
 	span = &types.Span{
 		TraceID: traceID3,
 		Event: types.Event{
-			Data:   map[string]interface{}{},
+			Data:   types.NewPayload(map[string]interface{}{}),
 			APIKey: legacyAPIKey,
 		},
 		IsRoot: true,
@@ -656,7 +656,7 @@ func TestDryRunMode(t *testing.T) {
 	// * remove the trace from the cache
 	events = transmission.GetBlock(1)
 	require.Equal(t, 1, len(events), "adding a root span should send the span")
-	assert.Equal(t, keepTraceID3, events[0].Data[config.DryRunFieldName], "field should match sampling decision for its trace ID")
+	assert.Equal(t, keepTraceID3, events[0].Data.Get(config.DryRunFieldName), "field should match sampling decision for its trace ID")
 	assert.Nil(t, coll.getFromCache(traceID3), "after sending the span, it should be removed from the cache")
 
 }
@@ -798,7 +798,7 @@ func TestStableMaxAlloc(t *testing.T) {
 			TraceID: strconv.Itoa(i),
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data:    spandata[i],
+				Data:    types.NewPayload(spandata[i]),
 				APIKey:  legacyAPIKey,
 			},
 		}
@@ -1005,18 +1005,18 @@ func TestAddCountsToRoot(t *testing.T) {
 			TraceID: traceID,
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "unused",
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 		}
 		switch i {
 		case 0, 1:
-			span.Data["meta.annotation_type"] = "span_event"
-			span.Data["meta.refinery.min_span"] = true
+			span.Data.Set("meta.annotation_type", "span_event")
+			span.Data.Set("meta.refinery.min_span", true)
 		case 2:
-			span.Data["meta.annotation_type"] = "link"
+			span.Data.Set("meta.annotation_type", "link")
 		}
 		coll.AddSpanFromPeer(span)
 	}
@@ -1028,7 +1028,7 @@ func TestAddCountsToRoot(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1037,15 +1037,15 @@ func TestAddCountsToRoot(t *testing.T) {
 
 	events := transmission.GetBlock(3)
 	assert.Equal(t, 3, len(events), "adding a root span should send all spans in the trace")
-	assert.Equal(t, nil, events[0].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, nil, events[1].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, nil, events[1].Data["meta.span_event_count"], "child span metadata should NOT be populated with span event count")
-	assert.Equal(t, nil, events[1].Data["meta.span_link_count"], "child span metadata should NOT be populated with span link count")
-	assert.Equal(t, nil, events[1].Data["meta.event_count"], "child span metadata should NOT be populated with event count")
-	assert.Equal(t, int64(2), events[2].Data["meta.span_count"], "root span metadata should be populated with span count")
-	assert.Equal(t, int64(2), events[2].Data["meta.span_event_count"], "root span metadata should be populated with span event count")
-	assert.Equal(t, int64(1), events[2].Data["meta.span_link_count"], "root span metadata should be populated with span link count")
-	assert.Equal(t, int64(5), events[2].Data["meta.event_count"], "root span metadata should be populated with event count")
+	assert.Equal(t, nil, events[0].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, nil, events[1].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, nil, events[1].Data.Get("meta.span_event_count"), "child span metadata should NOT be populated with span event count")
+	assert.Equal(t, nil, events[1].Data.Get("meta.span_link_count"), "child span metadata should NOT be populated with span link count")
+	assert.Equal(t, nil, events[1].Data.Get("meta.event_count"), "child span metadata should NOT be populated with event count")
+	assert.Equal(t, int64(2), events[2].Data.Get("meta.span_count"), "root span metadata should be populated with span count")
+	assert.Equal(t, int64(2), events[2].Data.Get("meta.span_event_count"), "root span metadata should be populated with span event count")
+	assert.Equal(t, int64(1), events[2].Data.Get("meta.span_link_count"), "root span metadata should be populated with span link count")
+	assert.Equal(t, int64(5), events[2].Data.Get("meta.event_count"), "root span metadata should be populated with event count")
 	assert.Nil(t, coll.getFromCache(traceID), "after adding a leaf and root span, it should be removed from the cache")
 }
 
@@ -1099,29 +1099,29 @@ func TestLateRootGetsCounts(t *testing.T) {
 			TraceID: traceID,
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "unused",
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 		}
 		switch i {
 		case 0, 1:
-			span.Data["meta.annotation_type"] = "span_event"
+			span.Data.Set("meta.annotation_type", "span_event")
 		case 2:
-			span.Data["meta.annotation_type"] = "link"
-			span.Data["meta.refinery.min_span"] = true
+			span.Data.Set("meta.annotation_type", "link")
+			span.Data.Set("meta.refinery.min_span", true)
 		}
 		coll.AddSpanFromPeer(span)
 	}
 
 	childSpans := transmission.GetBlock(3)
 	assert.Equal(t, 3, len(childSpans), "adding a non-root span and waiting should send the span")
-	assert.Equal(t, nil, childSpans[0].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, nil, childSpans[1].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, nil, childSpans[1].Data["meta.span_event_count"], "child span metadata should NOT be populated with span event count")
-	assert.Equal(t, nil, childSpans[1].Data["meta.span_link_count"], "child span metadata should NOT be populated with span link count")
-	assert.Equal(t, nil, childSpans[1].Data["meta.event_count"], "child span metadata should NOT be populated with event count")
+	assert.Equal(t, nil, childSpans[0].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, nil, childSpans[1].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, nil, childSpans[1].Data.Get("meta.span_event_count"), "child span metadata should NOT be populated with span event count")
+	assert.Equal(t, nil, childSpans[1].Data.Get("meta.span_link_count"), "child span metadata should NOT be populated with span link count")
+	assert.Equal(t, nil, childSpans[1].Data.Get("meta.event_count"), "child span metadata should NOT be populated with event count")
 	trace := coll.getFromCache(traceID)
 	assert.Nil(t, trace, "trace should have been sent although the root span hasn't arrived")
 
@@ -1130,7 +1130,7 @@ func TestLateRootGetsCounts(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1139,11 +1139,11 @@ func TestLateRootGetsCounts(t *testing.T) {
 
 	event := transmission.GetBlock(1)
 	assert.Equal(t, 1, len(event), "adding a root span should send all spans in the trace")
-	assert.Equal(t, int64(2), event[0].Data["meta.span_count"], "root span metadata should be populated with span count")
-	assert.Equal(t, int64(2), event[0].Data["meta.span_event_count"], "root span metadata should be populated with span event count")
-	assert.Equal(t, int64(1), event[0].Data["meta.span_link_count"], "root span metadata should be populated with span link count")
-	assert.Equal(t, int64(5), event[0].Data["meta.event_count"], "root span metadata should be populated with event count")
-	assert.Equal(t, "deterministic/always - late arriving span", event[0].Data["meta.refinery.reason"], "late spans should have meta.refinery.reason set to rules + late arriving span.")
+	assert.Equal(t, int64(2), event[0].Data.Get("meta.span_count"), "root span metadata should be populated with span count")
+	assert.Equal(t, int64(2), event[0].Data.Get("meta.span_event_count"), "root span metadata should be populated with span event count")
+	assert.Equal(t, int64(1), event[0].Data.Get("meta.span_link_count"), "root span metadata should be populated with span link count")
+	assert.Equal(t, int64(5), event[0].Data.Get("meta.event_count"), "root span metadata should be populated with event count")
+	assert.Equal(t, "deterministic/always - late arriving span", event[0].Data.Get("meta.refinery.reason"), "late spans should have meta.refinery.reason set to rules + late arriving span.")
 	assert.Nil(t, coll.getFromCache(traceID), "after adding a leaf and root span, it should be removed from the cache")
 }
 
@@ -1195,9 +1195,9 @@ func TestAddSpanCount(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1205,10 +1205,10 @@ func TestAddSpanCount(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id":        "unused",
 				"meta.refinery.min_span": true,
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1225,7 +1225,7 @@ func TestAddSpanCount(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1234,8 +1234,8 @@ func TestAddSpanCount(t *testing.T) {
 
 	events := transmission.GetBlock(2)
 	assert.Equal(t, 2, len(events), "adding a root span should send all spans in the trace")
-	assert.Equal(t, nil, events[0].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, int64(3), events[1].Data["meta.span_count"], "root span metadata should be populated with span count")
+	assert.Equal(t, nil, events[0].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, int64(3), events[1].Data.Get("meta.span_count"), "root span metadata should be populated with span count")
 	assert.Nil(t, coll.getFromCache(traceID), "after adding a leaf and root span, it should be removed from the cache")
 }
 
@@ -1287,9 +1287,9 @@ func TestLateRootGetsSpanCount(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1297,14 +1297,14 @@ func TestLateRootGetsSpanCount(t *testing.T) {
 
 	childSpans := transmission.GetBlock(1)
 	assert.Equal(t, 1, len(childSpans), "adding a non-root span and waiting should send the span")
-	assert.Equal(t, nil, childSpans[0].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
+	assert.Equal(t, nil, childSpans[0].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
 
 	// now we add the root span and verify that both got sent and that the root span had the span count
 	rootSpan := &types.Span{
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1313,8 +1313,8 @@ func TestLateRootGetsSpanCount(t *testing.T) {
 
 	events := transmission.GetBlock(1)
 	assert.Equal(t, 1, len(events), "adding a root span should send all spans in the trace")
-	assert.Equal(t, int64(2), events[0].Data["meta.span_count"], "root span metadata should be populated with span count")
-	assert.Equal(t, "deterministic/always - late arriving span", events[0].Data["meta.refinery.reason"], "late spans should have meta.refinery.reason set to late.")
+	assert.Equal(t, int64(2), events[0].Data.Get("meta.span_count"), "root span metadata should be populated with span count")
+	assert.Equal(t, "deterministic/always - late arriving span", events[0].Data.Get("meta.refinery.reason"), "late spans should have meta.refinery.reason set to late.")
 
 	assert.Nil(t, coll.getFromCache(traceID), "after adding a leaf and root span, it should be removed from the cache")
 }
@@ -1366,9 +1366,9 @@ func TestLateSpanNotDecorated(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1378,7 +1378,7 @@ func TestLateSpanNotDecorated(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1388,7 +1388,7 @@ func TestLateSpanNotDecorated(t *testing.T) {
 	events := transmission.GetBlock(2)
 	assert.Equal(t, 2, len(events), "adding a root span should send all spans in the trace")
 	if len(events) == 2 {
-		assert.Equal(t, nil, events[1].Data["meta.refinery.reason"], "late span should not have meta.refinery.reason set to late")
+		assert.Equal(t, nil, events[1].Data.Get("meta.refinery.reason"), "late span should not have meta.refinery.reason set to late")
 	}
 }
 
@@ -1439,9 +1439,9 @@ func TestAddAdditionalAttributes(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1452,7 +1452,7 @@ func TestAddAdditionalAttributes(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1461,8 +1461,8 @@ func TestAddAdditionalAttributes(t *testing.T) {
 
 	events := transmission.GetBlock(2)
 	assert.Equal(t, 2, len(events), "should be some events transmitted")
-	assert.Equal(t, "foo", events[0].Data["name"], "new attribute should appear in data")
-	assert.Equal(t, "bar", events[0].Data["other"], "new attribute should appear in data")
+	assert.Equal(t, "foo", events[0].Data.Get("name"), "new attribute should appear in data")
+	assert.Equal(t, "bar", events[0].Data.Get("other"), "new attribute should appear in data")
 }
 
 func TestStressReliefSampleRate(t *testing.T) {
@@ -1498,9 +1498,9 @@ func TestStressReliefSampleRate(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1527,7 +1527,7 @@ func TestStressReliefSampleRate(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset:    "aoeu",
-			Data:       map[string]interface{}{},
+			Data:       types.NewPayload(map[string]interface{}{}),
 			APIKey:     legacyAPIKey,
 			SampleRate: 10,
 		},
@@ -1601,9 +1601,9 @@ func TestStressReliefDecorateHostname(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id": "unused",
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -1613,7 +1613,7 @@ func TestStressReliefDecorateHostname(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -1622,7 +1622,7 @@ func TestStressReliefDecorateHostname(t *testing.T) {
 
 	events := transmission.GetBlock(2)
 	assert.Equal(t, 2, len(events), "adding a root span should send all spans in the trace")
-	assert.Equal(t, "host123", events[1].Data["meta.refinery.local_hostname"])
+	assert.Equal(t, "host123", events[1].Data.Get("meta.refinery.local_hostname"))
 
 }
 
@@ -1709,20 +1709,20 @@ func TestSpanWithRuleReasons(t *testing.T) {
 		span := &types.Span{
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id":  "unused",
 					"http.status_code": 200,
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 		}
 		switch i {
 		case 0, 1:
 			span.TraceID = traceIDs[0]
-			span.Data["test"] = int64(1)
+			span.Data.Set("test", int64(1))
 		case 2, 3:
 			span.TraceID = traceIDs[1]
-			span.Data["test"] = int64(2)
+			span.Data.Set("test", int64(2))
 		}
 		coll.AddSpanFromPeer(span)
 	}
@@ -1730,8 +1730,8 @@ func TestSpanWithRuleReasons(t *testing.T) {
 	eventsWithoutRoot := transmission.GetBlock(4)
 	assert.Equal(t, 4, len(eventsWithoutRoot), "traces should have been sent due to trace timeout")
 	for _, event := range eventsWithoutRoot {
-		reason := event.Data["meta.refinery.reason"]
-		if event.Data["test"] == int64(1) {
+		reason := event.Data.Get("meta.refinery.reason")
+		if event.Data.Get("test") == int64(1) {
 			assert.Equal(t, "rules/trace/rule 1:dynamic", reason, event.Data)
 		} else {
 			assert.Equal(t, "rules/span/rule 2:emadynamic", reason, event.Data)
@@ -1744,17 +1744,17 @@ func TestSpanWithRuleReasons(t *testing.T) {
 			TraceID: traceID,
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": 200,
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 			IsRoot: true,
 		}
 		if i == 0 {
-			rootSpan.Data["test"] = int64(1)
+			rootSpan.Data.Set("test", int64(1))
 		} else {
-			rootSpan.Data["test"] = int64(2)
+			rootSpan.Data.Set("test", int64(2))
 		}
 
 		coll.AddSpan(rootSpan)
@@ -1764,8 +1764,8 @@ func TestSpanWithRuleReasons(t *testing.T) {
 	roots := transmission.GetBlock(2)
 	assert.Equal(t, 2, len(roots), "root span should be sent immediately")
 	for _, event := range roots {
-		reason := event.Data["meta.refinery.reason"]
-		if event.Data["test"] == int64(1) {
+		reason := event.Data.Get("meta.refinery.reason")
+		if event.Data.Get("test") == int64(1) {
 			assert.Equal(t, "rules/trace/rule 1:dynamic - late arriving span", reason, event.Data)
 		} else {
 			assert.Equal(t, "rules/span/rule 2:emadynamic - late arriving span", reason, event.Data)
@@ -1834,7 +1834,7 @@ func TestRedistributeTraces(t *testing.T) {
 		Event: types.Event{
 			Dataset: dataset,
 			APIKey:  legacyAPIKey,
-			Data:    make(map[string]interface{}),
+			Data:    types.NewPayload(make(map[string]interface{})),
 		},
 	}
 
@@ -1873,7 +1873,7 @@ func TestRedistributeTraces(t *testing.T) {
 			Dataset: dataset,
 			APIKey:  legacyAPIKey,
 			APIHost: "api1",
-			Data:    make(map[string]interface{}),
+			Data:    types.NewPayload(make(map[string]interface{})),
 		},
 		IsRoot: true,
 	}
@@ -1882,10 +1882,9 @@ func TestRedistributeTraces(t *testing.T) {
 		Event: types.Event{
 			Dataset: dataset,
 			APIKey:  legacyAPIKey,
-			Data: map[string]interface {
-			}{
+			Data: types.NewPayload(map[string]interface{}{
 				"meta.refinery.min_span": true,
-			},
+			}),
 		},
 	}
 
@@ -2042,7 +2041,7 @@ func TestDrainTracesOnShutdown(t *testing.T) {
 		{
 			name:                 "Trace already has decision, should be sent",
 			traceID:              "traceID1",
-			span:                 &types.Span{TraceID: "traceID1", Event: types.Event{Dataset: "aoeu", Data: make(map[string]interface{})}},
+			span:                 &types.Span{TraceID: "traceID1", Event: types.Event{Dataset: "aoeu", Data: types.NewPayload(make(map[string]interface{}))}},
 			preRecordTrace:       true,
 			expectedSent:         1,
 			expectedForwarded:    0,
@@ -2051,7 +2050,7 @@ func TestDrainTracesOnShutdown(t *testing.T) {
 		{
 			name:                 "Trace cannot be decided yet, should be forwarded",
 			traceID:              "traceID2",
-			span:                 &types.Span{TraceID: "traceID2", Event: types.Event{Dataset: "test2", Data: make(map[string]interface{})}},
+			span:                 &types.Span{TraceID: "traceID2", Event: types.Event{Dataset: "test2", Data: types.NewPayload(make(map[string]interface{}))}},
 			preRecordTrace:       false,
 			expectedSent:         0,
 			expectedForwarded:    1,
@@ -2062,9 +2061,9 @@ func TestDrainTracesOnShutdown(t *testing.T) {
 		{
 			name:    "decision spans that already has decision should be ignored and discarded",
 			traceID: "traceID3",
-			span: &types.Span{TraceID: "traceID3", Event: types.Event{Dataset: "test3", Data: map[string]interface{}{
+			span: &types.Span{TraceID: "traceID3", Event: types.Event{Dataset: "test3", Data: types.NewPayload(map[string]interface{}{
 				"meta.refinery.min_span": true,
-			}}},
+			})}},
 			preRecordTrace:       true,
 			expectedSent:         0,
 			expectedForwarded:    0,
@@ -2073,9 +2072,9 @@ func TestDrainTracesOnShutdown(t *testing.T) {
 		{
 			name:    "decision spans that belongs to other peers should be ignored and discarded",
 			traceID: "traceID2",
-			span: &types.Span{TraceID: "traceID2", Event: types.Event{Dataset: "test4", Data: map[string]interface{}{
+			span: &types.Span{TraceID: "traceID2", Event: types.Event{Dataset: "test4", Data: types.NewPayload(map[string]interface{}{
 				"meta.refinery.min_span": true,
-			}}},
+			})}},
 			preRecordTrace:       false,
 			expectedSent:         0,
 			expectedForwarded:    0,
@@ -2111,12 +2110,12 @@ func TestDrainTracesOnShutdown(t *testing.T) {
 			require.Len(t, events, 1)
 			require.Equal(t, tt.span.Dataset, events[0].Dataset)
 
-			sendBy, ok := events[0].Data["meta.refinery.send_by"]
+			sendBy := events[0].Data.Get("meta.refinery.send_by")
 			if tt.expectedTraceSendBy != 0 {
-				require.True(t, ok)
+				require.NotNil(t, sendBy)
 				require.Equal(t, tt.expectedTraceSendBy, sendBy.(int64))
 			} else {
-				require.False(t, ok)
+				require.Nil(t, sendBy)
 			}
 
 			if tt.expectedAPIHost != "" {
@@ -2178,10 +2177,10 @@ func TestBigTracesGoEarly(t *testing.T) {
 			TraceID: traceID,
 			Event: types.Event{
 				Dataset: "aoeu",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "unused",
 					"index":           i,
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 		}
@@ -2197,7 +2196,7 @@ func TestBigTracesGoEarly(t *testing.T) {
 		TraceID: traceID,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data:    map[string]interface{}{},
+			Data:    types.NewPayload(map[string]interface{}{}),
 			APIKey:  legacyAPIKey,
 		},
 		IsRoot: true,
@@ -2206,13 +2205,13 @@ func TestBigTracesGoEarly(t *testing.T) {
 
 	rootEvents := transmission.GetBlock(1)
 	assert.Equal(t, 1, len(rootEvents), "hitting the spanlimit should send the trace")
-	assert.Equal(t, nil, childEvents[0].Data["meta.span_count"], "child span metadata should NOT be populated with span count")
-	assert.Equal(t, "trace_send_span_limit", childEvents[0].Data["meta.refinery.send_reason"], "child span metadata should set to trace_send_span_limit")
-	assert.EqualValues(t, spanlimit+1, rootEvents[0].Data["meta.span_count"], "root span metadata should be populated with span count")
-	assert.EqualValues(t, spanlimit+1, rootEvents[0].Data["meta.event_count"], "root span metadata should be populated with event count")
-	assert.Equal(t, "deterministic/chance - late arriving span", rootEvents[0].Data["meta.refinery.reason"], "the late root span should have meta.refinery.reason set to rules + late arriving span.")
+	assert.Equal(t, nil, childEvents[0].Data.Get("meta.span_count"), "child span metadata should NOT be populated with span count")
+	assert.Equal(t, "trace_send_span_limit", childEvents[0].Data.Get("meta.refinery.send_reason"), "child span metadata should set to trace_send_span_limit")
+	assert.EqualValues(t, spanlimit+1, rootEvents[0].Data.Get("meta.span_count"), "root span metadata should be populated with span count")
+	assert.EqualValues(t, spanlimit+1, rootEvents[0].Data.Get("meta.event_count"), "root span metadata should be populated with event count")
+	assert.Equal(t, "deterministic/chance - late arriving span", rootEvents[0].Data.Get("meta.refinery.reason"), "the late root span should have meta.refinery.reason set to rules + late arriving span.")
 	assert.EqualValues(t, 2, rootEvents[0].SampleRate, "the late root span should sample rate set")
-	assert.Equal(t, "trace_send_late_span", rootEvents[0].Data["meta.refinery.send_reason"], "send reason should indicate span count exceeded")
+	assert.Equal(t, "trace_send_late_span", rootEvents[0].Data.Get("meta.refinery.send_reason"), "send reason should indicate span count exceeded")
 
 }
 
@@ -2253,12 +2252,12 @@ func TestCreateDecisionSpan(t *testing.T) {
 		TraceID: traceID1,
 		Event: types.Event{
 			Dataset: "aoeu",
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.parent_id":        "unused",
 				"http.status_code":       200,
 				"test":                   1,
 				"should-not-be-included": 123,
-			},
+			}),
 			APIKey: legacyAPIKey,
 		},
 	}
@@ -2274,7 +2273,7 @@ func TestCreateDecisionSpan(t *testing.T) {
 		Dataset: "aoeu",
 		APIHost: peerShard.Addr,
 		APIKey:  legacyAPIKey,
-		Data: map[string]interface{}{
+		Data: types.NewPayload(map[string]interface{}{
 			"meta.annotation_type":         types.SpanAnnotationTypeUnknown,
 			"meta.refinery.min_span":       true,
 			"meta.refinery.root":           false,
@@ -2282,7 +2281,7 @@ func TestCreateDecisionSpan(t *testing.T) {
 			"meta.trace_id":                traceID1,
 			"http.status_code":             200,
 			"test":                         1,
-		},
+		}),
 	}
 
 	assert.Equal(t, expected.APIHost, ds.APIHost)
@@ -2297,7 +2296,7 @@ func TestCreateDecisionSpan(t *testing.T) {
 	rootSpan.IsRoot = true
 
 	ds = coll.createDecisionSpan(rootSpan, trace, peerShard)
-	expected.Data["meta.refinery.root"] = true
+	expected.Data.Set("meta.refinery.root", true)
 
 	assert.Equal(t, expected.APIHost, ds.APIHost)
 	assert.Equal(t, expected.APIKey, ds.APIKey)
@@ -2457,7 +2456,7 @@ func TestExpiredTracesCleanup(t *testing.T) {
 
 	events := peerTransmission.GetBlock(3)
 	assert.Len(t, events, 3)
-	assert.NotEmpty(t, events[0].Data["meta.refinery.expired_trace"])
+	assert.NotEmpty(t, events[0].Data.Get("meta.refinery.expired_trace"))
 
 	coll.sendExpiredTracesInCache(context.Background(), coll.Clock.Now().Add(5*traceTimeout))
 
@@ -2529,9 +2528,9 @@ func TestSpanLimitSendByPreservation(t *testing.T) {
 			TraceID: traceID,
 			Event: types.Event{
 				Dataset: "test-dataset",
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "unused",
-				},
+				}),
 				APIKey: legacyAPIKey,
 			},
 		})
@@ -2658,17 +2657,17 @@ func BenchmarkCollectorWithSamplers(b *testing.B) {
 							Event: types.Event{
 								Dataset: "benchmark-dataset",
 								APIKey:  "test-api-key",
-								Data: map[string]interface{}{
+								Data: types.NewPayload(map[string]interface{}{
 									"sampler-field-1": rand.Intn(20),
 									"sampler-field-2": "static-value",
 									"index":           spanIdx,
-								},
+								}),
 							},
 						}
 
 						// Add parent ID to non-root spans
 						if !isRoot {
-							spans[spanIdx].Data["trace.parent_id"] = fmt.Sprintf("parent-%s", traceID)
+							spans[spanIdx].Data.Set("trace.parent_id", fmt.Sprintf("parent-%s", traceID))
 						}
 
 						spanIdx++
diff --git a/route/otlp_logs_test.go b/route/otlp_logs_test.go
index 145bcc2..852c18b 100644
--- a/route/otlp_logs_test.go
+++ b/route/otlp_logs_test.go
@@ -428,7 +428,7 @@ func TestLogsOTLPHandler(t *testing.T) {
 		assert.Equal(t, 1, len(events))
 
 		event := events[0]
-		assert.Equal(t, "my-user-agent", event.Data["meta.refinery.incoming_user_agent"])
+		assert.Equal(t, "my-user-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 
 		assert.Equal(t, 0, len(router.Collector.(*collect.MockCollector).Spans))
 		mockCollector.Flush()
@@ -461,7 +461,7 @@ func TestLogsOTLPHandler(t *testing.T) {
 		events := mockTransmission.GetBlock(1)
 		assert.Equal(t, 1, len(events))
 		event := events[0]
-		assert.Equal(t, "my-user-agent", event.Data["meta.refinery.incoming_user_agent"])
+		assert.Equal(t, "my-user-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 
 		assert.Equal(t, 0, len(router.Collector.(*collect.MockCollector).Spans))
 		mockCollector.Flush()
diff --git a/route/otlp_trace_test.go b/route/otlp_trace_test.go
index 7d46d73..47e5451 100644
--- a/route/otlp_trace_test.go
+++ b/route/otlp_trace_test.go
@@ -149,12 +149,12 @@ func TestOTLPHandler(t *testing.T) {
 
 		spanEvent := events[0]
 		// assert.Equal(t, time.Unix(0, int64(12345)).UTC(), spanEvent.Timestamp)
-		assert.Equal(t, huskyotlp.BytesToTraceID(traceID), spanEvent.Data["trace.trace_id"])
-		assert.Equal(t, hex.EncodeToString(spanID), spanEvent.Data["trace.span_id"])
-		assert.Equal(t, "span_link", spanEvent.Data["span.name"])
-		assert.Equal(t, "span_with_event", spanEvent.Data["parent.name"])
-		assert.Equal(t, "span_event", spanEvent.Data["meta.annotation_type"])
-		assert.Equal(t, "event_attr_key", spanEvent.Data["event_attr_val"])
+		assert.Equal(t, huskyotlp.BytesToTraceID(traceID), spanEvent.Data.Get("trace.trace_id"))
+		assert.Equal(t, hex.EncodeToString(spanID), spanEvent.Data.Get("trace.span_id"))
+		assert.Equal(t, "span_link", spanEvent.Data.Get("span.name"))
+		assert.Equal(t, "span_with_event", spanEvent.Data.Get("parent.name"))
+		assert.Equal(t, "span_event", spanEvent.Data.Get("meta.annotation_type"))
+		assert.Equal(t, "event_attr_key", spanEvent.Data.Get("event_attr_val"))
 	})
 
 	t.Run("creates events for span links", func(t *testing.T) {
@@ -194,12 +194,12 @@ func TestOTLPHandler(t *testing.T) {
 		assert.Equal(t, 2, len(events))
 
 		spanLink := events[1]
-		assert.Equal(t, huskyotlp.BytesToTraceID(traceID), spanLink.Data["trace.trace_id"])
-		assert.Equal(t, hex.EncodeToString(spanID), spanLink.Data["trace.span_id"])
-		assert.Equal(t, huskyotlp.BytesToTraceID(linkTraceID), spanLink.Data["trace.link.trace_id"])
-		assert.Equal(t, hex.EncodeToString(linkSpanID), spanLink.Data["trace.link.span_id"])
-		assert.Equal(t, "link", spanLink.Data["meta.annotation_type"])
-		assert.Equal(t, "link_attr_val", spanLink.Data["link_attr_key"])
+		assert.Equal(t, huskyotlp.BytesToTraceID(traceID), spanLink.Data.Get("trace.trace_id"))
+		assert.Equal(t, hex.EncodeToString(spanID), spanLink.Data.Get("trace.span_id"))
+		assert.Equal(t, huskyotlp.BytesToTraceID(linkTraceID), spanLink.Data.Get("trace.link.trace_id"))
+		assert.Equal(t, hex.EncodeToString(linkSpanID), spanLink.Data.Get("trace.link.span_id"))
+		assert.Equal(t, "link", spanLink.Data.Get("meta.annotation_type"))
+		assert.Equal(t, "link_attr_val", spanLink.Data.Get("link_attr_key"))
 	})
 
 	t.Run("invalid headers", func(t *testing.T) {
@@ -552,7 +552,7 @@ func TestOTLPHandler(t *testing.T) {
 		assert.Equal(t, 2, len(events))
 
 		event := events[0]
-		assert.Equal(t, "my-user-agent", event.Data["meta.refinery.incoming_user_agent"])
+		assert.Equal(t, "my-user-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 	})
 
 	t.Run("spans record incoming user agent - HTTP", func(t *testing.T) {
@@ -582,7 +582,7 @@ func TestOTLPHandler(t *testing.T) {
 		assert.Equal(t, 2, len(events))
 
 		event := events[0]
-		assert.Equal(t, "my-user-agent", event.Data["meta.refinery.incoming_user_agent"])
+		assert.Equal(t, "my-user-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 		mockTransmission.Flush()
 	})
 
diff --git a/route/route_test.go b/route/route_test.go
index 4c774b7..b40f846 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -118,144 +119,162 @@ func TestDecompression(t *testing.T) {
 	}
 }
 
-func unmarshalRequest(w *httptest.ResponseRecorder, content string, body io.Reader) {
-	http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		var data map[string]interface{}
-		err := unmarshal(r, r.Body, &data)
-		if err != nil {
-			w.WriteHeader(http.StatusBadRequest)
-			w.Write([]byte(err.Error()))
-			return
-		}
-
-		var traceID string
-		if trID, ok := data["trace.trace_id"]; ok {
-			traceID = trID.(string)
-		} else if trID, ok := data["traceId"]; ok {
-			traceID = trID.(string)
-		}
-
-		w.Write([]byte(traceID))
-	}).ServeHTTP(w, &http.Request{
-		Body: io.NopCloser(body),
-		Header: http.Header{
-			"Content-Type": []string{content},
-		},
-	})
-}
-
-func unmarshalBatchRequest(w *httptest.ResponseRecorder, content string, body io.Reader) {
-	http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		var e batchedEvent
-		err := unmarshal(r, r.Body, &e)
-
-		if err != nil {
-			w.Write([]byte(err.Error()))
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-
-		w.Write([]byte(e.getEventTime().Format(time.RFC3339Nano)))
-	}).ServeHTTP(w, &http.Request{
-		Body: io.NopCloser(body),
-		Header: http.Header{
-			"Content-Type": []string{content},
-		},
-	})
-}
-
 func TestUnmarshal(t *testing.T) {
-	var w *httptest.ResponseRecorder
-	var body io.Reader
 	now := time.Now().UTC()
 
-	w = httptest.NewRecorder()
-	body = bytes.NewBufferString("")
-	unmarshalRequest(w, "nope", body)
-
-	if w.Code != http.StatusBadRequest {
-		t.Error("Expecting", http.StatusBadRequest, "Received", w.Code)
+	// Common test data - using only floats to avoid JSON type conversion issues
+	testData := map[string]interface{}{
+		"trace.trace_id": "test-trace-id",
+		"trace.span_id":  "test-span-id",
+		"service.name":   "test-service",
+		"operation.name": "test-operation",
+		"duration_ms":    150.5,
+		"status_code":    200.0,
+		"user_id":        12345.0,
+		"is_error":       false,
 	}
 
-	w = httptest.NewRecorder()
-	body = bytes.NewBufferString(`{"trace.trace_id": "test"}`)
-	unmarshalRequest(w, "application/json", body)
+	t.Run("invalid content type defaults to JSON", func(t *testing.T) {
+		req := httptest.NewRequest("POST", "/test", bytes.NewBufferString("{}"))
+		req.Header.Set("Content-Type", "nope")
 
-	if b := w.Body.String(); b != "test" {
-		t.Error("Expecting test")
-	}
+		var data map[string]interface{}
+		err := unmarshal(req, req.Body, &data)
+		// Should succeed because invalid content type defaults to JSON
+		assert.NoError(t, err)
+	})
 
-	w = httptest.NewRecorder()
-	body = bytes.NewBufferString(`{"traceId": "test"}`)
-	unmarshalRequest(w, "application/json; charset=utf-8", body)
+	// Test map[string]interface{} unmarshaling (used in requestToEvent)
+	t.Run("map[string]interface{}", func(t *testing.T) {
+		t.Run("json", func(t *testing.T) {
+			jsonData, err := json.Marshal(testData)
+			require.NoError(t, err)
 
-	if b := w.Body.String(); b != "test" {
-		t.Error("Expecting test")
-	}
+			for _, contentType := range []string{"application/json", "application/json; charset=utf-8"} {
+				t.Run(contentType, func(t *testing.T) {
+					req := httptest.NewRequest("POST", "/test", bytes.NewReader(jsonData))
+					req.Header.Set("Content-Type", contentType)
 
-	w = httptest.NewRecorder()
-	body = bytes.NewBufferString(fmt.Sprintf(`{"time": "%s"}`, now.Format(time.RFC3339Nano)))
-	unmarshalBatchRequest(w, "application/json", body)
+					var result map[string]interface{}
+					err = unmarshal(req, req.Body, &result)
+					require.NoError(t, err)
 
-	if b := w.Body.String(); b != now.Format(time.RFC3339Nano) {
-		t.Error("Expecting", now, "Received", b)
-	}
+					// Compare directly to test data
+					assert.Equal(t, testData, result)
+				})
+			}
+		})
 
-	var buf *bytes.Buffer
-	var e *msgpack.Encoder
-	var in map[string]interface{}
-	var err error
+		t.Run("msgpack", func(t *testing.T) {
+			for _, contentType := range []string{"application/msgpack", "application/x-msgpack"} {
+				t.Run(contentType, func(t *testing.T) {
+					buf := &bytes.Buffer{}
+					encoder := msgpack.NewEncoder(buf)
+					err := encoder.Encode(testData)
+					require.NoError(t, err)
 
-	w = httptest.NewRecorder()
-	buf = &bytes.Buffer{}
-	e = msgpack.NewEncoder(buf)
-	in = map[string]interface{}{"trace.trace_id": "test"}
-	err = e.Encode(in)
+					req := httptest.NewRequest("POST", "/test", buf)
+					req.Header.Set("Content-Type", contentType)
 
-	if err != nil {
-		t.Error(err)
-	}
+					var result map[string]interface{}
+					err = unmarshal(req, req.Body, &result)
+					require.NoError(t, err)
 
-	body = buf
-	unmarshalRequest(w, "application/msgpack", body)
+					// Compare directly to test data
+					assert.Equal(t, testData, result)
+				})
+			}
+		})
+	})
 
-	if b := w.Body.String(); b != "test" {
-		t.Error("Expecting test")
-	}
+	// Test []batchedEvent unmarshaling (used in batch)
+	t.Run("[]batchedEvent", func(t *testing.T) {
+		t.Run("json", func(t *testing.T) {
+			batchEvents := []batchedEvent{
+				{
+					Timestamp:  now.Format(time.RFC3339Nano),
+					SampleRate: 2,
+					Data:       types.NewPayload(testData),
+				},
+				{
+					Timestamp:  now.Add(time.Second).Format(time.RFC3339Nano),
+					SampleRate: 4,
+					Data:       types.NewPayload(testData),
+				},
+			}
+			jsonData, err := json.Marshal(batchEvents)
+			require.NoError(t, err)
+
+			for _, contentType := range []string{"application/json", "application/json; charset=utf-8"} {
+				t.Run(contentType, func(t *testing.T) {
+					req := httptest.NewRequest("POST", "/test", bytes.NewReader(jsonData))
+					req.Header.Set("Content-Type", contentType)
+
+					var result []batchedEvent
+					err = unmarshal(req, req.Body, &result)
+					require.NoError(t, err)
+					require.Len(t, result, 2)
+
+					assert.Equal(t, now.UTC(), result[0].getEventTime())
+					assert.Equal(t, uint(2), result[0].getSampleRate())
+					assert.Equal(t, testData, maps.Collect(result[0].Data.All()))
+
+					assert.Equal(t, now.Add(time.Second).UTC(), result[1].getEventTime())
+					assert.Equal(t, uint(4), result[1].getSampleRate())
+					assert.Equal(t, testData, maps.Collect(result[1].Data.All()))
+				})
+			}
+		})
 
-	w = httptest.NewRecorder()
-	buf = &bytes.Buffer{}
-	e = msgpack.NewEncoder(buf)
-	in = map[string]interface{}{"traceId": "test"}
-	err = e.Encode(in)
+		t.Run("msgpack", func(t *testing.T) {
+			// Create test data as a simple struct that can be marshaled/unmarshaled
+			type testBatchEvent struct {
+				MsgPackTimestamp *time.Time             `msgpack:"time,omitempty"`
+				SampleRate       int64                  `msgpack:"samplerate"`
+				Data             map[string]interface{} `msgpack:"data"`
+			}
 
-	if err != nil {
-		t.Error(err)
-	}
+			later := now.Add(time.Second)
+			testEvents := []testBatchEvent{
+				{
+					MsgPackTimestamp: &now,
+					SampleRate:       3,
+					Data:             testData,
+				},
+				{
+					MsgPackTimestamp: &later,
+					SampleRate:       6,
+					Data:             testData,
+				},
+			}
 
-	body = buf
-	unmarshalRequest(w, "application/msgpack", body)
+			for _, contentType := range []string{"application/msgpack", "application/x-msgpack"} {
+				t.Run(contentType, func(t *testing.T) {
+					buf := &bytes.Buffer{}
+					encoder := msgpack.NewEncoder(buf)
+					err := encoder.Encode(testEvents)
+					require.NoError(t, err)
 
-	if b := w.Body.String(); b != "test" {
-		t.Error("Expecting test")
-	}
+					req := httptest.NewRequest("POST", "/test", buf)
+					req.Header.Set("Content-Type", contentType)
 
-	w = httptest.NewRecorder()
-	buf = &bytes.Buffer{}
-	e = msgpack.NewEncoder(buf)
-	in = map[string]interface{}{"time": now}
-	err = e.Encode(in)
+					var result []batchedEvent
+					err = unmarshal(req, req.Body, &result)
+					require.NoError(t, err)
+					require.Len(t, result, 2)
 
-	if err != nil {
-		t.Error(err)
-	}
+					assert.Equal(t, now.UTC(), result[0].getEventTime())
+					assert.Equal(t, uint(3), result[0].getSampleRate())
+					assert.Equal(t, testData, maps.Collect(result[0].Data.All()))
 
-	body = buf
-	unmarshalBatchRequest(w, "application/msgpack", body)
+					assert.Equal(t, now.Add(time.Second).UTC(), result[1].getEventTime())
+					assert.Equal(t, uint(6), result[1].getSampleRate())
+					assert.Equal(t, testData, maps.Collect(result[1].Data.All()))
 
-	if b := w.Body.String(); b != now.Format(time.RFC3339Nano) {
-		t.Error("Expecting", now, "Received", b)
-	}
+				})
+			}
+		})
+	})
 }
 
 func TestGetAPIKeyAndDatasetFromMetadataCaseInsensitive(t *testing.T) {
@@ -736,34 +755,34 @@ func TestIsRootSpan(t *testing.T) {
 		{
 			name: "root span - no parent id",
 			event: types.Event{
-				Data: map[string]interface{}{},
+				Data: types.NewPayload(map[string]interface{}{}),
 			},
 			expected: true,
 		},
 		{
 			name: "root span - empty parent id",
 			event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "",
-				},
+				}),
 			},
 			expected: true,
 		},
 		{
 			name: "non-root span - parent id",
 			event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.parent_id": "some-id",
-				},
+				}),
 			},
 			expected: false,
 		},
 		{
 			name: "non-root span - no parent id but has signal_type of log",
 			event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"meta.signal_type": "log",
-				},
+				}),
 			},
 			expected: false,
 		},
@@ -783,22 +802,22 @@ func TestIsRootSpan(t *testing.T) {
 func TestAddIncomingUserAgent(t *testing.T) {
 	t.Run("no incoming user agent", func(t *testing.T) {
 		event := &types.Event{
-			Data: map[string]interface{}{},
+			Data: types.NewPayload(map[string]interface{}{}),
 		}
 
 		addIncomingUserAgent(event, "test-agent")
-		require.Equal(t, "test-agent", event.Data["meta.refinery.incoming_user_agent"])
+		require.Equal(t, "test-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 	})
 
 	t.Run("existing incoming user agent", func(t *testing.T) {
 		event := &types.Event{
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"meta.refinery.incoming_user_agent": "test-agent",
-			},
+			}),
 		}
 
 		addIncomingUserAgent(event, "another-test-agent")
-		require.Equal(t, "test-agent", event.Data["meta.refinery.incoming_user_agent"])
+		require.Equal(t, "test-agent", event.Data.Get("meta.refinery.incoming_user_agent"))
 	})
 }
 
@@ -900,12 +919,12 @@ func TestProcessEventMetrics(t *testing.T) {
 				APIHost:   "test.honeycomb.io",
 				Dataset:   "test-dataset",
 				Timestamp: time.Now(),
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"trace.trace_id":    "trace-123",
 					"meta.signal_type":  tt.signalType,
 					"test_attribute":    "test_value",
 					"another_attribute": 123,
-				},
+				}),
 			}
 			span := &types.Span{
 				Event:   *event,
@@ -970,7 +989,7 @@ func createBatchEvents() []batchedEvent {
 		{
 			Timestamp:  now.Format(time.RFC3339Nano),
 			SampleRate: 2,
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.trace_id":  "trace-1",
 				"trace.span_id":   "span-1",
 				"trace.parent_id": "",
@@ -979,12 +998,12 @@ func createBatchEvents() []batchedEvent {
 				"duration_ms":     100.0,
 				"int.field":       int64(1),
 				"bool.field":      true,
-			},
+			}),
 		},
 		{
 			Timestamp:  now.Format(time.RFC3339Nano),
 			SampleRate: 2,
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.trace_id":  "trace-1",
 				"trace.span_id":   "span-2",
 				"trace.parent_id": "span-1",
@@ -993,12 +1012,12 @@ func createBatchEvents() []batchedEvent {
 				"duration_ms":     50.0,
 				"int.field":       int64(2),
 				"bool.field":      false,
-			},
+			}),
 		},
 		{
 			Timestamp:  now.Format(time.RFC3339Nano),
 			SampleRate: 4,
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"trace.trace_id":  "trace-2",
 				"trace.span_id":   "span-3",
 				"trace.parent_id": "",
@@ -1006,7 +1025,7 @@ func createBatchEvents() []batchedEvent {
 				"operation.name":  "another-operation",
 				"duration_ms":     200.0,
 				"int.field":       int64(3),
-			},
+			}),
 		},
 	}
 	return batchEvents
@@ -1062,9 +1081,12 @@ func TestRouterBatch(t *testing.T) {
 
 	assert.Len(t, spans, len(batchEvents))
 	for i, span := range spans {
-		assert.Equal(t, batchEvents[i].Data["trace.trace_id"], span.TraceID)
+		assert.Equal(t, batchEvents[i].Data.Get("trace.trace_id"), span.TraceID)
 		assert.Equal(t, uint(batchEvents[i].SampleRate), span.SampleRate)
-		assert.Equal(t, batchEvents[i].Data, span.Data)
+		// Compare data values
+		for k, v := range batchEvents[i].Data.All() {
+			assert.Equal(t, v, span.Data.Get(k), "Data field %s should match", k)
+		}
 	}
 }
 
diff --git a/sample/dynamic_ema_test.go b/sample/dynamic_ema_test.go
index d344071..43302b6 100644
--- a/sample/dynamic_ema_test.go
+++ b/sample/dynamic_ema_test.go
@@ -29,12 +29,12 @@ func TestDynamicEMAAddSampleRateKeyToTrace(t *testing.T) {
 	for i := 0; i < spanCount; i++ {
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": 200,
 					"app.team.id":      float64(4),
 					"important_field":  true,
 					"request.path":     "/{slug}/fun",
-				},
+				}),
 			},
 		})
 	}
diff --git a/sample/dynamic_test.go b/sample/dynamic_test.go
index 358d8d6..aa9c7bf 100644
--- a/sample/dynamic_test.go
+++ b/sample/dynamic_test.go
@@ -31,19 +31,19 @@ func TestDynamicAddSampleRateKeyToTrace(t *testing.T) {
 		if i == spanCount-1 {
 			trace.RootSpan = &types.Span{
 				Event: types.Event{
-					Data: map[string]interface{}{
+					Data: types.NewPayload(map[string]interface{}{
 						"http.status_code": "200",
 						"service_name":     "test",
-					},
+					}),
 				},
 			}
 		}
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": "200",
 					"url":              "/test",
-				},
+				}),
 			},
 		})
 	}
diff --git a/sample/ema_throughput_test.go b/sample/ema_throughput_test.go
index 152e439..4803de3 100644
--- a/sample/ema_throughput_test.go
+++ b/sample/ema_throughput_test.go
@@ -29,12 +29,12 @@ func TestEMAThroughputAddSampleRateKeyToTrace(t *testing.T) {
 	for i := 0; i < spanCount; i++ {
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": 200,
 					"app.team.id":      float64(4),
 					"important_field":  true,
 					"request.path":     "/{slug}/fun",
-				},
+				}),
 			},
 		})
 	}
diff --git a/sample/rules_test.go b/sample/rules_test.go
index b9d70be..f27a46d 100644
--- a/sample/rules_test.go
+++ b/sample/rules_test.go
@@ -45,9 +45,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -74,9 +74,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(2),
-						},
+						}),
 					},
 				},
 			},
@@ -103,9 +103,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -132,9 +132,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -165,9 +165,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -205,17 +205,17 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":     2.2,
 							"test_two": false,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test_two": true,
-						},
+						}),
 					},
 				},
 			},
@@ -242,9 +242,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": float64(3),
-						},
+						}),
 					},
 				},
 			},
@@ -264,9 +264,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -298,16 +298,16 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": int64(1),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -336,9 +336,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": int64(9),
-						},
+						}),
 					},
 				},
 			},
@@ -364,9 +364,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": int64(9),
-						},
+						}),
 					},
 				},
 			},
@@ -392,9 +392,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"second": int64(9),
-						},
+						}),
 					},
 				},
 			},
@@ -421,9 +421,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": "honeycomb",
-						},
+						}),
 					},
 				},
 			},
@@ -450,9 +450,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": "honeycomb",
-						},
+						}),
 					},
 				},
 			},
@@ -479,9 +479,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"first": "honeycomb",
-						},
+						}),
 					},
 				},
 			},
@@ -508,9 +508,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(2),
-						},
+						}),
 					},
 				},
 			},
@@ -537,22 +537,22 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "54321",
 							"meta.span_count": int64(2),
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54321",
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 			},
@@ -581,32 +581,32 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "54321",
 							"meta.span_count": int64(2),
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54321",
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54321",
 							"test":            int64(3),
-						},
+						}),
 					},
 				},
 			},
@@ -634,22 +634,22 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54321",
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54321",
 							"test":            int64(3),
-						},
+						}),
 					},
 				},
 			},
@@ -676,32 +676,32 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "54321",
 							"meta.span_count": int64(2),
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54321",
 							"test":            int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54321",
 							"test":            int64(3),
-						},
+						}),
 					},
 				},
 			},
@@ -729,9 +729,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test2": int64(17),
-						},
+						}),
 					},
 				},
 			},
@@ -758,9 +758,9 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test2": int64(16),
-						},
+						}),
 					},
 				},
 			},
@@ -788,10 +788,10 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":  int64(2),
 							"test2": int64(17),
-						},
+						}),
 					},
 				},
 			},
@@ -821,39 +821,39 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "54322",
 							"trace.parent_id": "54321",
 							"meta.span_count": int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 			},
@@ -881,39 +881,39 @@ func TestRules(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "54322",
 							"trace.parent_id": "54321",
 							"meta.span_count": int64(2),
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "654321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"trace.trace_id":  "12345",
 							"trace.span_id":   "754321",
 							"trace.parent_id": "54322",
-						},
+						}),
 					},
 				},
 			},
@@ -941,7 +941,7 @@ func TestRules(t *testing.T) {
 
 		for _, span := range d.Spans {
 			trace.AddSpan(span)
-			if _, ok := span.Data["trace.parent_id"]; ok == false {
+			if span.Data.Get("trace.parent_id") == nil {
 				trace.RootSpan = span
 			}
 		}
@@ -969,6 +969,8 @@ func TestRules(t *testing.T) {
 }
 
 func TestRulesWithNestedFields(t *testing.T) {
+	// TODO test this with serialized msgpack as well, we want to make sure the behavior is equivalent.
+
 	data := []TestRulesData{
 		{
 			Rules: &config.RulesBasedSamplerConfig{
@@ -990,11 +992,11 @@ func TestRulesWithNestedFields(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": map[string]interface{}{
 								"test1": "a",
 							},
-						},
+						}),
 					},
 				},
 			},
@@ -1022,9 +1024,9 @@ func TestRulesWithNestedFields(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test.test1": "a",
-						},
+						}),
 					},
 				},
 			},
@@ -1051,11 +1053,11 @@ func TestRulesWithNestedFields(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": map[string]interface{}{
 								"test2": "b",
 							},
-						},
+						}),
 					},
 				},
 			},
@@ -1082,11 +1084,11 @@ func TestRulesWithNestedFields(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": map[string]interface{}{
 								"test1": "a",
 							},
-						},
+						}),
 					},
 				},
 			},
@@ -1115,11 +1117,11 @@ func TestRulesWithNestedFields(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": map[string]interface{}{
 								"test2": "a",
 							},
-						},
+						}),
 					},
 				},
 			},
@@ -1197,18 +1199,18 @@ func TestRulesWithDynamicSampler(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 			},
@@ -1288,18 +1290,18 @@ func TestRulesWithEMADynamicSampler(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 			},
@@ -1366,18 +1368,18 @@ func TestRuleMatchesSpanMatchingSpan(t *testing.T) {
 			spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(5),
 							"http.status_code": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -1389,18 +1391,18 @@ func TestRuleMatchesSpanMatchingSpan(t *testing.T) {
 			spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "500",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(5),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 			},
@@ -1491,9 +1493,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(1),
-						},
+						}),
 					},
 				},
 			},
@@ -1520,9 +1522,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": float64(1.01),
-						},
+						}),
 					},
 				},
 			},
@@ -1549,9 +1551,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -1578,9 +1580,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": true,
-						},
+						}),
 					},
 				},
 			},
@@ -1607,9 +1609,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": true,
-						},
+						}),
 					},
 				},
 			},
@@ -1636,9 +1638,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": true,
-						},
+						}),
 					},
 				},
 			},
@@ -1665,9 +1667,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": 10.,
-						},
+						}),
 					},
 				},
 			},
@@ -1694,9 +1696,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": 100,
-						},
+						}),
 					},
 				},
 			},
@@ -1723,9 +1725,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -1752,9 +1754,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -1781,9 +1783,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": 2,
-						},
+						}),
 					},
 				},
 			},
@@ -1810,9 +1812,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": 9.3, // "9.3" is greater than "10.3" when compared as a string
-						},
+						}),
 					},
 				},
 			},
@@ -1839,9 +1841,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "4.13",
-						},
+						}),
 					},
 				},
 			},
@@ -1868,9 +1870,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "fourPointOneThree",
-						},
+						}),
 					},
 				},
 			},
@@ -1897,9 +1899,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "rightValue",
-						},
+						}),
 					},
 				},
 			},
@@ -1926,9 +1928,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": 777,
-						},
+						}),
 					},
 				},
 			},
@@ -1955,9 +1957,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(11),
-						},
+						}),
 					},
 				},
 			},
@@ -1983,9 +1985,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": int64(123),
-						},
+						}),
 					},
 				},
 			},
@@ -2011,9 +2013,9 @@ func TestRulesDatatypes(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "abc",
-						},
+						}),
 					},
 				},
 			},
@@ -2104,9 +2106,9 @@ func TestRegexpRules(t *testing.T) {
 			spans := []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": d.value,
-						},
+						}),
 					},
 				},
 			}
@@ -2146,18 +2148,18 @@ func TestRulesWithDeterministicSampler(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"rule_test":        int64(1),
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 			},
@@ -2227,25 +2229,25 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "foo",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test1": 1,
 							"test2": 2.2,
 							"test3": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -2278,25 +2280,25 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "foo",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test1": 1,
 							"test2": 2.2,
 							"test3": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -2328,27 +2330,27 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":         "foo",
 							"anotherField": 3.5,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
 							"anotherField":     1.1,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test1": 1,
 							"test2": 2.2,
 							"test3": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -2384,27 +2386,27 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":         "foo",
 							"anotherField": 50,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
 							"anotherField":     10,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test1": 1,
 							"test2": 2.2,
 							"test3": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -2440,27 +2442,27 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":         "foo",
 							"anotherField": 50,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
 							"anotherField":     10,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test1": 1,
 							"test2": 2.2,
 							"test3": "foo",
-						},
+						}),
 					},
 				},
 			},
@@ -2489,25 +2491,25 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "nope", // I am the root span that does not match.
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":  "foo", // I am the span that almost matches, but I'm not root.
 							"test1": 1,
 							"test2": 2.2,
-						},
+						}),
 					},
 				},
 			},
@@ -2536,25 +2538,25 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "nope",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test":  "nope",
 							"test1": 1,
 							"test2": 2.2,
-						},
+						}),
 					},
 				},
 			},
@@ -2582,23 +2584,23 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "nope",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -2626,23 +2628,23 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "123testABC", // I am root.
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"test": "nope",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -2669,16 +2671,16 @@ func TestRulesRootSpanContext(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "200",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"http.status_code": "500",
-						},
+						}),
 					},
 				},
 			},
@@ -2705,17 +2707,17 @@ func TestRulesRootSpanContext(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": 100.01,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": 99.1,
 							"bar": 100.10,
-						},
+						}),
 					},
 				},
 			},
@@ -2742,16 +2744,16 @@ func TestRulesRootSpanContext(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": true,
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": true,
-						},
+						}),
 					},
 				},
 			},
@@ -2782,16 +2784,16 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "abc123",
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"name.of.service": "is not service.name!",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": true,
-						},
+						}),
 					},
 				},
 			},
@@ -2822,16 +2824,16 @@ func TestRulesRootSpanContext(t *testing.T) {
 				{
 					TraceID: "abc123",
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"service.name": "totally present",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": true,
-						},
+						}),
 					},
 				},
 			},
@@ -2858,16 +2860,16 @@ func TestRulesRootSpanContext(t *testing.T) {
 			Spans: []*types.Span{
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"service.name": "no trace id on this test span, so it's not root",
-						},
+						}),
 					},
 				},
 				{
 					Event: types.Event{
-						Data: map[string]interface{}{
+						Data: types.NewPayload(map[string]interface{}{
 							"foo": true,
-						},
+						}),
 					},
 				},
 			},
diff --git a/sample/totalthroughput_test.go b/sample/totalthroughput_test.go
index c8990bc..1d20b73 100644
--- a/sample/totalthroughput_test.go
+++ b/sample/totalthroughput_test.go
@@ -29,9 +29,9 @@ func TestTotalThroughputAddSampleRateKeyToTrace(t *testing.T) {
 	for i := 0; i < spanCount; i++ {
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": "200",
-				},
+				}),
 			},
 		})
 	}
diff --git a/sample/trace_key_test.go b/sample/trace_key_test.go
index 47f04d5..188e5df 100644
--- a/sample/trace_key_test.go
+++ b/sample/trace_key_test.go
@@ -119,19 +119,19 @@ func TestKeyLimits(t *testing.T) {
 	for i := 0; i < 160; i++ {
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"fieldA": fmt.Sprintf("value%d", i),
 					"fieldB": i,
-				},
+				}),
 			},
 		})
 	}
 
 	trace.RootSpan = &types.Span{
 		Event: types.Event{
-			Data: map[string]interface{}{
+			Data: types.NewPayload(map[string]interface{}{
 				"service_name": "test",
-			},
+			}),
 		},
 	}
 
@@ -444,7 +444,7 @@ func BenchmarkTraceKeyBuild(b *testing.B) {
 
 				span := &types.Span{
 					Event: types.Event{
-						Data: spanData,
+						Data: types.NewPayload(spanData),
 					},
 				}
 
@@ -457,7 +457,7 @@ func BenchmarkTraceKeyBuild(b *testing.B) {
 					for _, field := range scenario.fields {
 						if strings.HasPrefix(field, "root.") {
 							actualField := field[len("root."):]
-							span.Data[actualField] = fmt.Sprintf("root-%s-value", actualField)
+							span.Data.Set(actualField, fmt.Sprintf("root-%s-value", actualField))
 						}
 					}
 				}
@@ -486,7 +486,7 @@ func createTestTrace(t *testing.T, spans []testSpan) *types.Trace {
 	for _, s := range spans {
 		span := &types.Span{
 			Event: types.Event{
-				Data: s.data,
+				Data: types.NewPayload(s.data),
 			},
 		}
 		if s.isRoot {
diff --git a/sample/windowed_throughput_test.go b/sample/windowed_throughput_test.go
index ec81bb6..3746de6 100644
--- a/sample/windowed_throughput_test.go
+++ b/sample/windowed_throughput_test.go
@@ -29,12 +29,12 @@ func TestWindowedThroughputAddSampleRateKeyToTrace(t *testing.T) {
 	for i := 0; i < spanCount; i++ {
 		trace.AddSpan(&types.Span{
 			Event: types.Event{
-				Data: map[string]interface{}{
+				Data: types.NewPayload(map[string]interface{}{
 					"http.status_code": 200,
 					"app.team.id":      float64(4),
 					"important_field":  true,
 					"request.path":     "/{slug}/fun",
-				},
+				}),
 			},
 		})
 	}
diff --git a/types/event_test.go b/types/event_test.go
index 3397d59..b6c9724 100644
--- a/types/event_test.go
+++ b/types/event_test.go
@@ -1,6 +1,7 @@
 package types
 
 import (
+	"maps"
 	"strconv"
 	"strings"
 	"testing"
@@ -25,18 +26,19 @@ func TestSpan_GetDataSize(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			data := make(map[string]any)
+			for i := 0; i < tt.numInts; i++ {
+				data[tt.name+"int"+strconv.Itoa(i)] = i
+			}
+			for i := 0; i < tt.numStrings; i++ {
+				data[tt.name+"str"+strconv.Itoa(i)] = strings.Repeat("x", i)
+			}
 			sp := &Span{
 				TraceID: tt.name,
 				Event: Event{
-					Data: make(map[string]any),
+					Data: NewPayload(data),
 				},
 			}
-			for i := 0; i < tt.numInts; i++ {
-				sp.Data[tt.name+"int"+strconv.Itoa(i)] = i
-			}
-			for i := 0; i < tt.numStrings; i++ {
-				sp.Data[tt.name+"str"+strconv.Itoa(i)] = strings.Repeat("x", i)
-			}
 			if got := sp.GetDataSize(); got != tt.want {
 				t.Errorf("Span.CalculateSize() = %v, want %v", got, tt.want)
 			}
@@ -57,16 +59,15 @@ func TestSpan_GetDataSizeSlice(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			sliceData := make([]any, tt.num)
+			for i := range tt.num {
+				sliceData[i] = i
+			}
 			sp := &Span{
 				Event: Event{
-					Data: make(map[string]any),
+					Data: NewPayload(map[string]any{"data": sliceData}),
 				},
 			}
-			data := make([]any, tt.num)
-			for i := range tt.num {
-				data[i] = i
-			}
-			sp.Data["data"] = data
 			if got := sp.GetDataSize(); got != tt.want {
 				t.Errorf("Span.CalculateSize() = %v, want %v", got, tt.want)
 			}
@@ -87,16 +88,15 @@ func TestSpan_GetDataSizeMap(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			mapData := make(map[string]any)
+			for i := range tt.num {
+				mapData[strconv.Itoa(i)] = i
+			}
 			sp := &Span{
 				Event: Event{
-					Data: make(map[string]any),
+					Data: NewPayload(map[string]any{"data": mapData}),
 				},
 			}
-			data := make(map[string]any)
-			for i := range tt.num {
-				data[strconv.Itoa(i)] = i
-			}
-			sp.Data["data"] = data
 			if got := sp.GetDataSize(); got != tt.want {
 				t.Errorf("Span.CalculateSize() = %v, want %v", got, tt.want)
 			}
@@ -118,7 +118,7 @@ func TestSpan_AnnotationType(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			sp := &Span{
 				Event: Event{
-					Data: tt.data,
+					Data: NewPayload(tt.data),
 				},
 			}
 			if got := sp.AnnotationType(); got != tt.want {
@@ -136,10 +136,10 @@ func TestSpan_ExtractDecisionContext(t *testing.T) {
 		Environment: "test-environment",
 		SampleRate:  5,
 		Timestamp:   time.Now(),
-		Data: map[string]interface{}{
+		Data: NewPayload(map[string]interface{}{
 			"test":                 "test",
 			"meta.annotation_type": "span_event",
-		},
+		}),
 	}
 	sp := &Span{
 		Event:       ev,
@@ -155,13 +155,15 @@ func TestSpan_ExtractDecisionContext(t *testing.T) {
 	assert.Equal(t, ev.Environment, got.Environment)
 	assert.Equal(t, ev.SampleRate, got.SampleRate)
 	assert.Equal(t, ev.Timestamp, got.Timestamp)
-	assert.Equal(t, map[string]interface{}{
+	expectedData := map[string]interface{}{
 		"meta.trace_id":                sp.TraceID,
 		"meta.refinery.root":           true,
 		"meta.refinery.min_span":       true,
 		"meta.annotation_type":         SpanAnnotationTypeSpanEvent,
 		"meta.refinery.span_data_size": 38,
-	}, got.Data)
+	}
+	actualData := maps.Collect(got.Data.All())
+	assert.Equal(t, expectedData, actualData)
 }
 
 func TestSpan_IsDecisionSpan(t *testing.T) {
@@ -182,7 +184,7 @@ func TestSpan_IsDecisionSpan(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			sp := &Span{
 				Event: Event{
-					Data: tt.data,
+					Data: NewPayload(tt.data),
 				},
 			}
 			got := sp.IsDecisionSpan()
@@ -197,16 +199,17 @@ func TestSpan_IsDecisionSpan(t *testing.T) {
 // ~10 microseconds. Since these happen once per span, when adding it to a trace,
 // we don't expect this to be a performance issue.
 func BenchmarkSpan_CalculateSizeSmall(b *testing.B) {
-	sp := &Span{
-		Event: Event{
-			Data: make(map[string]any),
-		},
-	}
+	data := make(map[string]any)
 	for i := 0; i < 10; i++ {
-		sp.Data["int"+strconv.Itoa(i)] = i
+		data["int"+strconv.Itoa(i)] = i
 	}
 	for i := 0; i < 10; i++ {
-		sp.Data["str"+strconv.Itoa(i)] = strings.Repeat("x", i)
+		data["str"+strconv.Itoa(i)] = strings.Repeat("x", i)
+	}
+	sp := &Span{
+		Event: Event{
+			Data: NewPayload(data),
+		},
 	}
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
@@ -215,16 +218,17 @@ func BenchmarkSpan_CalculateSizeSmall(b *testing.B) {
 }
 
 func BenchmarkSpan_CalculateSizeLarge(b *testing.B) {
-	sp := &Span{
-		Event: Event{
-			Data: make(map[string]any),
-		},
-	}
+	data := make(map[string]any)
 	for i := 0; i < 500; i++ {
-		sp.Data["int"+strconv.Itoa(i)] = i
+		data["int"+strconv.Itoa(i)] = i
 	}
 	for i := 0; i < 500; i++ {
-		sp.Data["str"+strconv.Itoa(i)] = strings.Repeat("x", i)
+		data["str"+strconv.Itoa(i)] = strings.Repeat("x", i)
+	}
+	sp := &Span{
+		Event: Event{
+			Data: NewPayload(data),
+		},
 	}
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
diff --git a/types/msgp_payload_map_test.go b/types/msgp_payload_map_test.go
new file mode 100644
index 0000000..9cff043
--- /dev/null
+++ b/types/msgp_payload_map_test.go
@@ -0,0 +1,329 @@
+package types
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+func TestMsgpPayloadMap_WorkingCases(t *testing.T) {
+	inputData := map[string]any{
+		"string_field":   "hello world",
+		"int_field":      int8(42),
+		"negative_int":   int64(-123),
+		"float_field":    float32(4.125),
+		"bool_true":      true,
+		"bool_false":     false,
+		"empty_string":   "",
+		"zero_int":       int(0),
+		"zero_float":     0.0,
+		"large_int":      int64(9223372036854775807),  // max int64
+		"small_int":      int64(-9223372036854775808), // min int64
+		"unicode_string": "café ñoño 🚀",
+		"nested_map":     map[string]any{"inner": "value"},
+		"array_field":    []any{int64(1), int64(2), int64(3)},
+		"nil_field":      nil,
+	}
+
+	// Serialize test data
+	encoded, err := msgpack.Marshal(inputData)
+	require.NoError(t, err)
+
+	// Create MsgpPayloadMap
+	payloadMap := NewMessagePackPayloadMap(encoded)
+
+	// Test iteration
+	iter, err := payloadMap.Iterate()
+	require.NoError(t, err)
+
+	gotData := make(map[string]any)
+	for {
+		key, typ, err := iter.NextKey()
+		if err == io.EOF {
+			break
+		}
+		require.NoError(t, err)
+
+		keyStr := string(key)
+		switch typ {
+		case FieldTypeString:
+			val, err := iter.ValueString()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[keyStr], val)
+			gotData[keyStr] = val
+		case FieldTypeInt64:
+			val, err := iter.ValueInt64()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[keyStr], val)
+			gotData[keyStr] = val
+		case FieldTypeFloat64:
+			val, err := iter.ValueFloat64()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[keyStr], val)
+			gotData[keyStr] = val
+		case FieldTypeBool:
+			val, err := iter.ValueBool()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[keyStr], val)
+			gotData[keyStr] = val
+		case FieldTypeOther:
+			val, err := iter.ValueAny()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[keyStr], val)
+			gotData[keyStr] = val
+		default:
+			assert.Fail(t, "Unexpected field type", "Unexpected field type %v for key %s", typ, keyStr)
+		}
+	}
+	assert.Len(t, gotData, len(inputData))
+
+	// Test skipping all values
+	iter, err = payloadMap.Iterate()
+	require.NoError(t, err)
+
+	skipCount := 0
+	for {
+		_, _, err := iter.NextKey()
+		if err == io.EOF {
+			break
+		}
+		require.NoError(t, err)
+		skipCount++
+		// Don't call any Value methods - test that skipping works
+	}
+	assert.Equal(t, len(inputData), skipCount)
+
+	// Test skipping some, but not all values
+	iter, err = payloadMap.Iterate()
+	require.NoError(t, err)
+
+	clear(gotData)
+	skipCount = 0
+	for {
+		key, typ, err := iter.NextKey()
+		if err == io.EOF {
+			break
+		}
+		require.NoError(t, err)
+
+		if typ == FieldTypeInt64 {
+			val, err := iter.ValueInt64()
+			require.NoError(t, err)
+			assert.EqualValues(t, inputData[string(key)], val)
+			gotData[string(key)] = val
+		} else {
+			skipCount++
+		}
+	}
+	assert.Equal(t, len(inputData), skipCount+len(gotData))
+
+	// Test empty map
+	emptyData := map[string]any{}
+	emptyEncoded, err := msgpack.Marshal(emptyData)
+	require.NoError(t, err)
+
+	emptyPayloadMap := NewMessagePackPayloadMap(emptyEncoded)
+	emptyIter, err := emptyPayloadMap.Iterate()
+	require.NoError(t, err)
+
+	_, _, err = emptyIter.NextKey()
+	assert.Equal(t, io.EOF, err)
+}
+
+func TestMsgpPayloadMap_ErrorCases(t *testing.T) {
+	// Test invalid msgpack data
+	invalidData := []byte{0xFF, 0xFF, 0xFF}
+	invalidPayloadMap := NewMessagePackPayloadMap(invalidData)
+
+	_, err := invalidPayloadMap.Iterate()
+	assert.Error(t, err)
+
+	// Test non-map msgpack data (array)
+	arrayData := []any{1, 2, 3}
+	arrayEncoded, err := msgpack.Marshal(arrayData)
+	require.NoError(t, err)
+
+	arrayPayloadMap := NewMessagePackPayloadMap(arrayEncoded)
+	_, err = arrayPayloadMap.Iterate()
+	assert.Error(t, err)
+
+	// Test empty data
+	emptyPayloadMap := NewMessagePackPayloadMap(nil)
+	_, err = emptyPayloadMap.Iterate()
+	assert.Error(t, err)
+
+	// Test truncated data, should fail on the second value.
+	validData := map[string]any{"key": "value"}
+	validEncoded, err := msgpack.Marshal(validData)
+	require.NoError(t, err)
+
+	truncatedData := validEncoded[:len(validEncoded)-2]
+	truncatedPayloadMap := NewMessagePackPayloadMap(truncatedData)
+	truncatedIter, err := truncatedPayloadMap.Iterate()
+	assert.NoError(t, err)
+	_, _, err = truncatedIter.NextKey()
+	assert.NoError(t, err)
+	_, _, err = truncatedIter.NextKey()
+	assert.Error(t, err)
+
+	// Test type mismatch errors by creating a map with specific types
+	// and attempting to read them as wrong types
+	inputData := map[string]any{
+		"string_field": "hello",
+		"int_field":    int64(42),
+		"float_field":  3.14,
+		"bool_field":   true,
+	}
+
+	encoded, err := msgpack.Marshal(inputData)
+	require.NoError(t, err)
+
+	payloadMap := NewMessagePackPayloadMap(encoded)
+	iter, err := payloadMap.Iterate()
+	require.NoError(t, err)
+
+	for {
+		_, typ, err := iter.NextKey()
+		if err == io.EOF {
+			break
+		}
+		require.NoError(t, err)
+
+		// Try to access with wrong types
+		switch typ {
+		case FieldTypeString:
+			// Try to read as int
+			_, err := iter.ValueInt64()
+			assert.Error(t, err)
+		case FieldTypeInt64:
+			// Try to read as string
+			_, err := iter.ValueString()
+			assert.Error(t, err)
+		case FieldTypeFloat64:
+			// Try to read as bool
+			_, err := iter.ValueBool()
+			assert.Error(t, err)
+		case FieldTypeBool:
+			// Try to read as float
+			_, err := iter.ValueFloat64()
+			assert.Error(t, err)
+		}
+	}
+
+	// Test calling Value methods without NextKey
+	simpleData := map[string]any{"test": "value"}
+	simpleEncoded, err := msgpack.Marshal(simpleData)
+	require.NoError(t, err)
+
+	freshPayloadMap := NewMessagePackPayloadMap(simpleEncoded)
+	freshIter, err := freshPayloadMap.Iterate()
+	require.NoError(t, err)
+
+	_, err = freshIter.ValueString()
+	assert.Error(t, err)
+}
+
+func BenchmarkMsgpPayloadMap(b *testing.B) {
+	testData := map[string]any{
+		"trace_id":        "abc123def456",
+		"span_id":         "span789",
+		"duration_ms":     int64(1500),
+		"error_rate":      0.05,
+		"success":         true,
+		"service_name":    "my-service",
+		"endpoint":        "/api/v1/users",
+		"status_code":     int64(200),
+		"bytes_sent":      int64(2048),
+		"response_time":   125.5,
+		"cache_hit":       false,
+		"user_id":         int64(12345),
+		"session_id":      "sess_abcdef123456",
+		"region":          "us-west-2",
+		"version":         "1.2.3",
+		"build_id":        int64(987654321),
+		"cpu_usage":       0.75,
+		"memory_mb":       int64(512),
+		"disk_io":         1024.0,
+		"network_latency": 25.3,
+	}
+
+	encoded, err := msgpack.Marshal(testData)
+	require.NoError(b, err)
+
+	b.Run("iter_values", func(b *testing.B) {
+		for b.Loop() {
+			payloadMap := NewMessagePackPayloadMap(encoded)
+			iter, _ := payloadMap.Iterate()
+
+			fieldCount := 0
+			for {
+				_, typ, err := iter.NextKey()
+				if err == io.EOF {
+					break
+				}
+
+				// Access the value based on type
+				switch typ {
+				case FieldTypeString:
+					_, _ = iter.ValueString()
+				case FieldTypeInt64:
+					_, _ = iter.ValueInt64()
+				case FieldTypeFloat64:
+					_, _ = iter.ValueFloat64()
+				case FieldTypeBool:
+					_, _ = iter.ValueBool()
+				case FieldTypeOther:
+					_, _ = iter.ValueAny()
+				}
+				fieldCount++
+			}
+
+			if fieldCount != len(testData) {
+				b.Fatal("field count mismatch")
+			}
+		}
+	})
+
+	b.Run("iter_novalues", func(b *testing.B) {
+		for b.Loop() {
+			payloadMap := NewMessagePackPayloadMap(encoded)
+			iter, _ := payloadMap.Iterate()
+
+			fieldCount := 0
+			for {
+				_, _, err := iter.NextKey()
+				if err == io.EOF {
+					break
+				}
+				fieldCount++
+				// Don't access values - just skip them
+			}
+
+			if fieldCount != len(testData) {
+				b.Fatal("field count mismatch")
+			}
+		}
+	})
+
+	b.Run("legacy", func(b *testing.B) {
+		for b.Loop() {
+			var result map[string]any
+			_ = msgpack.Unmarshal(encoded, &result)
+
+			// Iterate through the deserialized map
+			fieldCount := 0
+			for key, value := range result {
+				_ = key
+				_ = value
+				fieldCount++
+			}
+
+			if fieldCount != len(testData) {
+				b.Fatal("field count mismatch")
+			}
+		}
+	})
+}
diff --git a/types/payload_test.go b/types/payload_test.go
new file mode 100644
index 0000000..d7205c7
--- /dev/null
+++ b/types/payload_test.go
@@ -0,0 +1,163 @@
+package types
+
+import (
+	"encoding/json"
+	"fmt"
+	"maps"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+func TestPayload(t *testing.T) {
+	data := map[string]any{
+		"key1":   "value1",
+		"key2":   int64(42),
+		"key3":   3.14,
+		"key4":   true,
+		"keyNil": nil,
+	}
+
+	var ph Payload
+	doTest := func(t *testing.T) {
+		assert.True(t, ph.Exists("key1"))
+		assert.Equal(t, "value1", ph.Get("key1"))
+		assert.True(t, ph.Exists("key2"))
+		assert.Equal(t, int64(42), ph.Get("key2"))
+		assert.True(t, ph.Exists("keyNil"))
+		assert.Nil(t, ph.Get("keyNil"))
+		assert.False(t, ph.Exists("nonexistent"))
+		assert.Nil(t, ph.Get("nonexistent"))
+
+		ph.Set("key5", "newvalue")
+		assert.True(t, ph.Exists("key5"))
+		assert.Equal(t, "newvalue", ph.Get("key5"))
+
+		// Overwrite an existing value
+		ph.Set("key3", 4.13)
+		assert.True(t, ph.Exists("key3"))
+		assert.Equal(t, 4.13, ph.Get("key3"))
+
+		expected := maps.Clone(data)
+		expected["key3"] = 4.13
+		expected["key5"] = "newvalue"
+		found := maps.Collect(ph.All())
+		assert.Equal(t, expected, found)
+
+		asJSON, err := json.Marshal(ph)
+		require.NoError(t, err)
+
+		var fromJSON Payload
+		err = json.Unmarshal(asJSON, &fromJSON)
+		require.NoError(t, err)
+
+		// round-tripping through JSON turns our ints into floats
+		expectedFromJSON := maps.Collect(ph.All())
+		expectedFromJSON["key2"] = 42.0
+		assert.EqualValues(t, expectedFromJSON, maps.Collect(fromJSON.All()))
+	}
+
+	ph = NewPayload(data)
+	t.Run("from_map", doTest)
+
+	msgpData, err := msgpack.Marshal(data)
+	require.NoError(t, err)
+	err = msgpack.Unmarshal(msgpData, &ph)
+	require.NoError(t, err)
+	t.Run("from_msgp", doTest)
+}
+
+func BenchmarkPayload(b *testing.B) {
+	// Create test data with many fields
+	var keys []string
+
+	data := make(map[string]any)
+	for i := 0; i < 100; i++ {
+		keys = append(keys, fmt.Sprintf("key%d", i))
+		data[fmt.Sprintf("key%d", i)] = fmt.Sprintf("value%d", i)
+		data[fmt.Sprintf("num%d", i)] = int64(i)
+		data[fmt.Sprintf("float%d", i)] = float64(i) * 1.5
+		data[fmt.Sprintf("bool%d", i)] = i%2 == 0
+	}
+
+	// Create msgp data using msgpack.Marshal
+	msgpData, err := msgpack.Marshal(data)
+	require.NoError(b, err)
+
+	phMap := NewPayload(data)
+	var phMsgp Payload
+	err = msgpack.Unmarshal(msgpData, &phMsgp)
+	require.NoError(b, err)
+
+	b.Run("create_map", func(b *testing.B) {
+		for b.Loop() {
+			var m map[string]any
+			_ = msgpack.Unmarshal(msgpData, &m)
+			_ = NewPayload(m)
+		}
+	})
+
+	b.Run("create_msgp", func(b *testing.B) {
+		for b.Loop() {
+			var phMsgp Payload
+			_ = phMsgp.UnmarshalMsgpack(msgpData)
+		}
+	})
+
+	for _, num := range []int{1, 5, 10, 25} {
+		b.Run(fmt.Sprintf("get_map/%d", num), func(b *testing.B) {
+			modulo := len(keys) - num
+			for n := range b.N {
+				// Look at a variety of keys, since their (random) position in
+				// the serialized data has a big impact on lookup time.
+				offset := n % modulo
+				for _, key := range keys[offset : offset+num] {
+					_ = phMap.Get(key)
+				}
+			}
+		})
+
+		b.Run(fmt.Sprintf("get_msgp/%d", num), func(b *testing.B) {
+			modulo := len(keys) - num
+			for n := range b.N {
+				offset := n % modulo
+				for _, key := range keys[offset : offset+num] {
+					_ = phMsgp.Get(key)
+				}
+			}
+		})
+
+		b.Run(fmt.Sprintf("get_memo/%d", num), func(b *testing.B) {
+			modulo := len(keys) - num
+			for n := range b.N {
+				offset := n % modulo
+				var phMsgpMemo Payload
+				_ = phMsgpMemo.UnmarshalMsgpack(msgpData)
+				phMsgpMemo.MemoizeFields(keys[offset : offset+num]...)
+				for _, key := range keys[offset : offset+num] {
+					_ = phMsgpMemo.Get(key)
+				}
+			}
+		})
+	}
+
+	b.Run("iter_map", func(b *testing.B) {
+		for b.Loop() {
+			for k, v := range phMap.All() {
+				_ = k
+				_ = v
+			}
+		}
+	})
+
+	b.Run("iter_msgp", func(b *testing.B) {
+		for b.Loop() {
+			for k, v := range phMsgp.All() {
+				_ = k
+				_ = v
+			}
+		}
+	})
+}