data converters with compression

Author: vishwa-uber

cmd/samples/recipes/dataconverter/README.md

@@ -1,23 +1,33 @@
 # Data Converter Sample
 
-This sample workflow demonstrates how to use custom data converters in Cadence workflows. The data converter is responsible for serializing and deserializing workflow inputs, outputs, and activity parameters.
+This sample workflow demonstrates how to use custom data converters in Cadence workflows with compression capabilities. The data converter is responsible for serializing and deserializing workflow inputs, outputs, and activity parameters, with the added benefit of data compression to save storage space and bandwidth.
 
 ## Sample Description
 
-The sample implements a custom JSON data converter that:
+The sample implements a custom compressed JSON data converter that:
 - Serializes workflow inputs and activity parameters to JSON format
-- Deserializes workflow outputs and activity results from JSON format
-- Provides better control over data serialization compared to the default data converter
-- Can be extended to support custom serialization formats (e.g., Protocol Buffers, MessagePack)
+- Compresses the JSON data using gzip compression to reduce size
+- Decompresses and deserializes workflow outputs and activity results from JSON format
+- Provides significant storage and bandwidth savings for large payloads
+- Demonstrates advanced data converter patterns for production use cases
+- Shows real-time compression statistics and size comparisons
 
-The workflow takes a `MyPayload` struct as input, processes it through an activity, and returns the modified payload.
+The sample includes two workflows:
+1. **Simple Workflow**: Processes a basic `MyPayload` struct
+2. **Large Payload Workflow**: Processes a complex `LargePayload` with nested objects, arrays, and extensive data to demonstrate compression benefits
+
+All data is automatically compressed during serialization and decompressed during deserialization, with compression statistics displayed at runtime.
 
 ## Key Components
 
-- **Custom Data Converter**: `jsonDataConverter` implements the `encoded.DataConverter` interface
-- **Workflow**: `dataConverterWorkflow` demonstrates using custom data types with the converter
-- **Activity**: `dataConverterActivity` processes the input and returns modified data
-- **Test**: Includes unit tests to verify the data converter functionality
+- **Custom Data Converter**: `compressedJSONDataConverter` implements the `encoded.DataConverter` interface with gzip compression
+- **Simple Workflow**: `dataConverterWorkflow` demonstrates basic payload processing with compression
+- **Large Payload Workflow**: `largeDataConverterWorkflow` demonstrates processing complex data structures with compression
+- **Activities**: `dataConverterActivity` and `largeDataConverterActivity` process different payload types
+- **Large Payload Generator**: `CreateLargePayload()` creates realistic complex data for compression demonstration
+- **Compression Statistics**: `GetPayloadSizeInfo()` shows before/after compression metrics
+- **Tests**: Includes unit tests for both simple and large payload workflows
+- **Compression**: Automatic gzip compression/decompression for all workflow data
 
 ## Steps to Run Sample
 
@@ -33,18 +43,32 @@ The workflow takes a `MyPayload` struct as input, processes it through an activi
    ./bin/dataconverter -m trigger
    ```
 
-You should see logs showing the workflow input being processed through the activity and the final result being returned.
+You should see:
+- Compression statistics showing original vs compressed data sizes
+- Workflow logs showing the processing of large payloads
+- Activity execution logs with payload information
+- Final workflow completion with compression benefits noted
 
 ## Customization
 
-To use a different serialization format, you can implement your own data converter by:
-1. Creating a struct that implements the `encoded.DataConverter` interface
-2. Implementing the `ToData` method for serialization
-3. Implementing the `FromData` method for deserialization
-4. Registering the converter in the worker options
+To implement your own data converter with compression or other features:
+1. Create a struct that implements the `encoded.DataConverter` interface
+2. Implement the `ToData` method for serialization and compression
+3. Implement the `FromData` method for decompression and deserialization
+4. Register the converter in the worker options
 
 This pattern is useful when you need to:
+- Reduce storage costs and bandwidth usage with compression
 - Use specific serialization formats for performance or compatibility
 - Add encryption/decryption to workflow data
-- Implement custom compression for large payloads
-- Support legacy data formats 
+- Implement custom compression algorithms (LZ4, Snappy, etc.)
+- Support legacy data formats
+- Add data validation or transformation during serialization
+
+## Performance Benefits
+
+The compressed data converter provides:
+- **Storage Savings**: Typically 60-80% reduction in data size for JSON payloads
+- **Bandwidth Reduction**: Lower network transfer costs and faster data transmission
+- **Cost Optimization**: Reduced storage costs in Cadence history
+- **Scalability**: Better performance with large payloads 

cmd/samples/recipes/dataconverter/dataconverter.go

@@ -2,33 +2,65 @@ package main
 
 import (
 	"bytes"
+	"compress/gzip"
 	"encoding/json"
 	"fmt"
+	"io"
 	"reflect"
 
 	"go.uber.org/cadence/encoded"
 )
 
-type jsonDataConverter struct{}
+type compressedJSONDataConverter struct{}
 
-func NewJSONDataConverter() encoded.DataConverter {
-	return &jsonDataConverter{}
+func NewCompressedJSONDataConverter() encoded.DataConverter {
+	return &compressedJSONDataConverter{}
 }
 
-func (dc *jsonDataConverter) ToData(value ...interface{}) ([]byte, error) {
-	var buf bytes.Buffer
-	enc := json.NewEncoder(&buf)
+func (dc *compressedJSONDataConverter) ToData(value ...interface{}) ([]byte, error) {
+	// First, serialize to JSON
+	var jsonBuf bytes.Buffer
+	enc := json.NewEncoder(&jsonBuf)
 	for i, obj := range value {
 		err := enc.Encode(obj)
 		if err != nil {
 			return nil, fmt.Errorf("unable to encode argument: %d, %v, with error: %v", i, reflect.TypeOf(obj), err)
 		}
 	}
-	return buf.Bytes(), nil
+
+	// Then compress the JSON data
+	var compressedBuf bytes.Buffer
+	gzipWriter := gzip.NewWriter(&compressedBuf)
+
+	_, err := gzipWriter.Write(jsonBuf.Bytes())
+	if err != nil {
+		return nil, fmt.Errorf("unable to compress data: %v", err)
+	}
+
+	err = gzipWriter.Close()
+	if err != nil {
+		return nil, fmt.Errorf("unable to close gzip writer: %v", err)
+	}
+
+	return compressedBuf.Bytes(), nil
 }
 
-func (dc *jsonDataConverter) FromData(input []byte, valuePtr ...interface{}) error {
-	dec := json.NewDecoder(bytes.NewBuffer(input))
+func (dc *compressedJSONDataConverter) FromData(input []byte, valuePtr ...interface{}) error {
+	// First, decompress the data
+	gzipReader, err := gzip.NewReader(bytes.NewBuffer(input))
+	if err != nil {
+		return fmt.Errorf("unable to create gzip reader: %v", err)
+	}
+	defer gzipReader.Close()
+
+	// Read the decompressed JSON data
+	decompressedData, err := io.ReadAll(gzipReader)
+	if err != nil {
+		return fmt.Errorf("unable to decompress data: %v", err)
+	}
+
+	// Then deserialize from JSON
+	dec := json.NewDecoder(bytes.NewBuffer(decompressedData))
 	for i, obj := range valuePtr {
 		err := dec.Decode(obj)
 		if err != nil {
@@ -37,3 +69,15 @@ func (dc *jsonDataConverter) FromData(input []byte, valuePtr ...interface{}) err
 	}
 	return nil
 }
+
+// GetCompressionRatio returns the compression ratio for demonstration purposes
+func (dc *compressedJSONDataConverter) GetCompressionRatio(originalData []byte) (float64, error) {
+	// Simulate the compression process to calculate ratio
+	compressedData, err := dc.ToData(string(originalData))
+	if err != nil {
+		return 0, err
+	}
+
+	ratio := float64(len(compressedData)) / float64(len(originalData))
+	return ratio, nil
+}