Merge pull request #238 from mooseburgr/bytes-decimal-codec-cache

Fix bug in caching of "bytes.decimal" codec to include "precision.scale"

codec.go

@@ -566,11 +566,26 @@ func buildCodecForTypeDescribedByString(st map[string]*Codec, enclosingNamespace
 		isLogicalType = true
 		searchType = fmt.Sprintf("%s.%s", typeName, lt)
 	}
+
 	// NOTE: When codec already exists, return it. This includes both primitive and
 	// logicalType codecs added in NewCodec, and user-defined types, added while
 	// building the codec.
 	if cd, ok := st[searchType]; ok {
-		return cd, nil
+
+		// For "bytes.decimal" types verify that the scale and precision in this schema map match a cached codec before
+		// using the cached codec in favor of creating a new codec.
+		if searchType == "bytes.decimal" {
+
+			// Search the cached codecs for a "bytes.decimal" codec  with a "precision" and "scale" specified in the key,
+			// only if that matches return the cached codec. Otherwise, create a new codec for this "bytes.decimal".
+			decimalSearchType := fmt.Sprintf("bytes.decimal.%d.%d", int(schemaMap["precision"].(float64)), int(schemaMap["scale"].(float64)))
+			if cd2, ok := st[decimalSearchType]; ok {
+				return cd2, nil
+			}
+
+		} else {
+			return cd, nil
+		}
 	}
 
 	// Avro specification allows abbreviation of type name inside a namespace.

codec_test.go

@@ -300,3 +300,34 @@ func ExampleSingleItemDecoding() {
 	fmt.Println(datum)
 	// Output: 3
 }
+
+func Test_buildCodecForTypeDescribedByString_CacheRespectsPrecisionScale(t *testing.T) {
+	schemaMap := map[string]interface{}{
+		"type":        "bytes",
+		"logicalType": "decimal",
+		"precision":   float64(4),
+		"scale":       float64(2),
+	}
+	cachedCodecIdentifier := "preexisting-cached-coded"
+	cache := map[string]*Codec{
+		"bytes.decimal": nil, // precision.scale-agnostic codec
+		"bytes.decimal.4.2": {
+			schemaOriginal: cachedCodecIdentifier, // using field as identifier
+		},
+	}
+
+	// cached bytes.decimal codec with matching precision.scale is returned
+	cacheHit, err := buildCodecForTypeDescribedByString(cache, "", "bytes", schemaMap, nil)
+	ensureError(t, err) // ensure NO error
+	if cacheHit.schemaOriginal != cachedCodecIdentifier {
+		t.Errorf("GOT: %v; WANT: %v", cacheHit.schemaOriginal, cachedCodecIdentifier)
+	}
+
+	// cached codec with unmatching precision.scale is not returned
+	schemaMap["scale"] = float64(1)
+	cacheMiss, err := buildCodecForTypeDescribedByString(cache, "", "bytes", schemaMap, nil)
+	ensureError(t, err) // ensure NO error
+	if cacheMiss.schemaOriginal == cachedCodecIdentifier {
+		t.Errorf("GOT: %v; WANT: %v", cacheMiss.schemaOriginal, "!= "+cachedCodecIdentifier)
+	}
+}

logical_type.go

@@ -271,6 +271,11 @@ func makeDecimalBytesCodec(st map[string]*Codec, enclosingNamespace string, sche
 	if err != nil {
 		return nil, fmt.Errorf("Bytes ought to have valid name: %s", err)
 	}
+
+	// Add an additional cached codec for this "bytes.decimal" keyed also by "precision" and "scale"
+	decimalSearchType := fmt.Sprintf("bytes.decimal.%d.%d", precision, scale)
+	st[decimalSearchType] = c
+
 	c.binaryFromNative = decimalBytesFromNative(bytesBinaryFromNative, toSignedBytes, precision, scale)
 	c.textualFromNative = decimalBytesFromNative(bytesTextualFromNative, toSignedBytes, precision, scale)
 	c.nativeFromBinary = nativeFromDecimalBytes(bytesNativeFromBinary, precision, scale)