Add comprehensive protobuf support for reading & comparing protobuf data

README.md

@@ -10,14 +10,19 @@ This is particularly useful for data validation, migration testing, and ensuring
 
 ## Features
 
-- **Multiple Data Sources:** Supports reading from different sources, starting with CSV and JSON-Lines (`.jsonl`).
+- **Multiple Data Sources:** Supports reading from different sources, including CSV, JSON-Lines (`.jsonl`), and Protocol Buffers (Protobuf).
 - **Automatic Schema Detection:**
     - Infers the schema from a sample of the data.
     - Flattens nested JSON objects and arrays into a dot-notation format (e.g., `customer.address.city`).
     - Automatically detects data types: `numeric`, `string`, `boolean`, `date`, `datetime`, `timestamp`.
 - **Advanced String Parsing:**
     - Can detect and recursively parse JSON strings embedded within other file formats (e.g., a CSV field containing a JSON object).
     - Identifies field patterns using a library of built-in regex matchers and supports custom matchers.
+- **Protobuf Support:**
+    - Reads JSON-serialized protobuf messages (most common format for streaming data).
+    - Supports both `protobuf` and `proto` as source types for convenience.
+    - Handles nested protobuf messages with automatic field flattening.
+    - Compatible with protobuf messages exported to JSON format from various systems.
 - **Intelligent Date/Time Handling:**
     - Parses and compares `date`, `datetime`, and `timestamp` fields, even if their string formats differ between sources.
     - Supports timestamps with variable precision.
@@ -35,7 +40,7 @@ The tool is configured using two YAML files, one for each data source.
 **Example `config.yaml`:**
 ```yaml
 source:
-  # Type of the data source. Supported: csv, json
+  # Type of the data source. Supported: csv, json, protobuf (or proto)
   type: csv
   # Path to the source file.
   path: path/to/your/data.csv
@@ -50,6 +55,17 @@ source:
 #     ...
 ```
 
+**Example Protobuf config:**
+```yaml
+source:
+  type: protobuf  # or "proto" for short
+  path: path/to/your/data.jsonpb
+  parser_config:
+    json_in_string: false  # Usually not needed for protobuf JSON
+  sampler:
+    sample_size: 1000  # Number of records to sample for schema detection
+```
+
 ## Usage
 
 To run a comparison, use the `compare` command and provide the paths to the two configuration files.

go.mod

@@ -2,4 +2,6 @@ module data-comparator
 
 go 1.25
 
-require gopkg.in/yaml.v3 v3.0.1 // indirect
+require gopkg.in/yaml.v3 v3.0.1
+
+require google.golang.org/protobuf v1.36.9 // indirect

go.sum

@@ -1,3 +1,6 @@
+google.golang.org/protobuf v1.36.9 h1:w2gp2mA27hUeUzj9Ex9FBjsBm40zfaDtEWow293U7Iw=
+google.golang.org/protobuf v1.36.9/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/pkg/datareader/datareader.go

@@ -24,6 +24,8 @@ func New(cfg config.Source) (DataReader, error) {
 		return NewCSVReader(cfg)
 	case "json":
 		return NewJSONReader(cfg)
+	case "protobuf", "proto":
+		return NewProtobufReader(cfg)
 	default:
 		return nil, fmt.Errorf("unsupported source type: %s", cfg.Type)
 	}

internal/pkg/datareader/datareader_test.go

@@ -134,3 +134,108 @@ func TestReader_EOF(t *testing.T) {
 		t.Errorf("Expected io.EOF, got %v", err)
 	}
 }
+
+func TestProtobufReader_JSONFormat(t *testing.T) {
+	cfg := config.Source{
+		Type: "protobuf",
+		Path: "../../../testdata/testcase4_protobuf/source1.jsonpb",
+	}
+	reader, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	defer reader.Close()
+
+	// Read first record
+	rec, err := reader.Read()
+	if err != nil {
+		t.Fatalf("Read() error = %v", err)
+	}
+
+	// Check basic field
+	if userID, ok := rec["user_id"].(string); ok {
+		if userID != "user-001" {
+			t.Errorf("user_id read incorrectly, got %s, want %s", userID, "user-001")
+		}
+	} else {
+		t.Error("Field 'user_id' is not a string")
+	}
+
+	// Check nested field
+	if profile, ok := rec["profile"].(map[string]interface{}); ok {
+		if email, ok := profile["email"].(string); ok {
+			if email != "alice@example.com" {
+				t.Errorf("Nested field email read incorrectly, got %s, want %s", email, "alice@example.com")
+			}
+		} else {
+			t.Error("Nested field 'email' is not a string")
+		}
+	} else {
+		t.Error("Field 'profile' is not a map")
+	}
+
+	// Check array field
+	if activity, ok := rec["activity"].(map[string]interface{}); ok {
+		if devices, ok := activity["devices"].([]interface{}); ok {
+			if len(devices) != 2 {
+				t.Errorf("devices array length incorrect, got %d, want %d", len(devices), 2)
+			}
+			if devices[0].(string) != "mobile" {
+				t.Errorf("First device incorrect, got %s, want %s", devices[0], "mobile")
+			}
+		} else {
+			t.Error("Field 'devices' is not an array")
+		}
+	} else {
+		t.Error("Field 'activity' is not a map")
+	}
+}
+
+func TestProtobufReader_MultipleTypes(t *testing.T) {
+	// Test both "protobuf" and "proto" as valid type names
+	for _, sourceType := range []string{"protobuf", "proto"} {
+		t.Run(sourceType, func(t *testing.T) {
+			cfg := config.Source{
+				Type: sourceType,
+				Path: "../../../testdata/testcase4_protobuf/source1.jsonpb",
+			}
+			reader, err := New(cfg)
+			if err != nil {
+				t.Fatalf("New() error for type %s = %v", sourceType, err)
+			}
+			defer reader.Close()
+
+			// Just verify we can read one record without error
+			_, err = reader.Read()
+			if err != nil {
+				t.Fatalf("Read() error for type %s = %v", sourceType, err)
+			}
+		})
+	}
+}
+
+func TestProtobufReader_EOF(t *testing.T) {
+	cfg := config.Source{
+		Type: "protobuf",
+		Path: "../../../testdata/testcase4_protobuf/source1.jsonpb",
+	}
+	reader, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	defer reader.Close()
+
+	// Read all 5 records
+	for i := 0; i < 5; i++ {
+		_, err := reader.Read()
+		if err != nil {
+			t.Fatalf("Read() error on record %d: %v", i+1, err)
+		}
+	}
+
+	// The next read should return io.EOF
+	_, err = reader.Read()
+	if err != io.EOF {
+		t.Errorf("Expected io.EOF, got %v", err)
+	}
+}