AVRO-4358: [C] Match enum symbols by name in schema resolution

lang/c/src/resolved-reader.c

@@ -298,12 +298,10 @@ avro_resolved_reader_get_schema(const avro_value_iface_t *viface, const void *vs
 }
 
 
-static avro_resolved_reader_t *
-avro_resolved_reader_create(avro_schema_t wschema, avro_schema_t rschema)
+static void
+avro_resolved_reader_init_iface(avro_resolved_reader_t *self,
+				avro_schema_t wschema, avro_schema_t rschema)
 {
-	avro_resolved_reader_t  *self = (avro_resolved_reader_t  *) avro_new(avro_resolved_reader_t);
-	memset(self, 0, sizeof(avro_resolved_reader_t));
-
 	self->parent.incref_iface = avro_resolved_reader_incref_iface;
 	self->parent.decref_iface = avro_resolved_reader_decref_iface;
 	self->parent.incref = avro_resolved_reader_incref;
@@ -318,6 +316,14 @@ avro_resolved_reader_create(avro_schema_t wschema, avro_schema_t rschema)
 	self->calculate_size = avro_resolved_reader_calculate_size_;
 	self->free_iface = avro_resolved_reader_free_iface;
 	self->reset_wrappers = NULL;
+}
+
+static avro_resolved_reader_t *
+avro_resolved_reader_create(avro_schema_t wschema, avro_schema_t rschema)
+{
+	avro_resolved_reader_t  *self = (avro_resolved_reader_t  *) avro_new(avro_resolved_reader_t);
+	memset(self, 0, sizeof(avro_resolved_reader_t));
+	avro_resolved_reader_init_iface(self, wschema, rschema);
 	return self;
 }
 
@@ -1625,14 +1631,58 @@ try_array(memoize_state_t *state,
  * enum
  */
 
+typedef struct avro_resolved_enum_reader {
+	avro_resolved_reader_t  parent;
+	int  symbol_count;
+	/* Map writer index to reader index with -1 when no mapping */
+	int  *index_mapping;
+} avro_resolved_enum_reader_t;
+
+static void
+avro_resolved_enum_reader_free_iface(avro_resolved_reader_t *iface,
+				     st_table *freeing)
+{
+	AVRO_UNUSED(freeing);
+	avro_resolved_enum_reader_t  *eiface =
+	    container_of(iface, avro_resolved_enum_reader_t, parent);
+	if (eiface->index_mapping != NULL) {
+		avro_free(eiface->index_mapping,
+			  eiface->symbol_count * sizeof(int));
+	}
+	avro_schema_decref(iface->wschema);
+	avro_schema_decref(iface->rschema);
+	avro_freet(avro_resolved_enum_reader_t, eiface);
+}
+
 static int
 avro_resolved_reader_get_enum(const avro_value_iface_t *viface,
 			      const void *vself, int *val)
 {
-	AVRO_UNUSED(viface);
+	int  rval;
+	const avro_resolved_reader_t  *iface =
+	    container_of(viface, avro_resolved_reader_t, parent);
+	const avro_resolved_enum_reader_t  *eiface =
+	    container_of(iface, avro_resolved_enum_reader_t, parent);
 	const avro_value_t  *src = (const avro_value_t *) vself;
+	int  writer_val;
+
 	DEBUG("Getting enum from %p", src->self);
-	return avro_value_get_enum(src, val);
+	check(rval, avro_value_get_enum(src, &writer_val));
+
+	if (writer_val < 0 || writer_val >= eiface->symbol_count) {
+		avro_set_error("Invalid enum value %d for %s", writer_val,
+			       avro_schema_name(iface->wschema));
+		return EINVAL;
+	}
+	if (eiface->index_mapping[writer_val] < 0) {
+		avro_set_error("Symbol %s isn't in the reader's %s enum",
+			       avro_schema_enum_get(iface->wschema, writer_val),
+			       avro_schema_name(iface->wschema));
+		return EINVAL;
+	}
+
+	*val = eiface->index_mapping[writer_val];
+	return 0;
 }
 
 static avro_resolved_reader_t *
@@ -1649,11 +1699,35 @@ try_enum(memoize_state_t *state,
 		const char  *rname = avro_schema_name(rschema);
 
 		if (strcmp(wname, rname) == 0) {
-			avro_resolved_reader_t  *self =
-			    avro_resolved_reader_create(wschema, rschema);
-			avro_memoize_set(&state->mem, wschema, rschema, self);
-			self->parent.get_enum = avro_resolved_reader_get_enum;
-			return self;
+			int  symbol_count =
+			    avro_schema_enum_number_of_symbols(wschema);
+			int  *index_mapping =
+			    (int *) avro_calloc(symbol_count, sizeof(int));
+			int  wi;
+
+			for (wi = 0; wi < symbol_count; wi++) {
+				const char  *symbol =
+				    avro_schema_enum_get(wschema, wi);
+				index_mapping[wi] = avro_schema_enum_get_by_name
+				    (rschema, symbol);
+			}
+
+			avro_resolved_enum_reader_t  *eself =
+			    (avro_resolved_enum_reader_t *)
+			    avro_new(avro_resolved_enum_reader_t);
+			memset(eself, 0, sizeof(avro_resolved_enum_reader_t));
+			avro_resolved_reader_init_iface(&eself->parent,
+							wschema, rschema);
+			eself->parent.free_iface =
+			    avro_resolved_enum_reader_free_iface;
+			eself->parent.parent.get_enum =
+			    avro_resolved_reader_get_enum;
+			eself->symbol_count = symbol_count;
+			eself->index_mapping = index_mapping;
+
+			avro_memoize_set(&state->mem, wschema, rschema,
+					 &eself->parent);
+			return &eself->parent;
 		}
 	}
 	avro_set_error("Writer %s not compatible with reader %s",

lang/c/src/resolved-writer.c

@@ -307,12 +307,10 @@ avro_resolved_writer_get_schema(const avro_value_iface_t *viface, const void *vs
 }
 
 
-static avro_resolved_writer_t *
-avro_resolved_writer_create(avro_schema_t wschema, avro_schema_t rschema)
+static void
+avro_resolved_writer_init_iface(avro_resolved_writer_t *self,
+				avro_schema_t wschema, avro_schema_t rschema)
 {
-	avro_resolved_writer_t  *self = (avro_resolved_writer_t *) avro_new(avro_resolved_writer_t);
-	memset(self, 0, sizeof(avro_resolved_writer_t));
-
 	self->parent.incref_iface = avro_resolved_writer_incref_iface;
 	self->parent.decref_iface = avro_resolved_writer_decref_iface;
 	self->parent.incref = avro_resolved_writer_incref;
@@ -328,6 +326,14 @@ avro_resolved_writer_create(avro_schema_t wschema, avro_schema_t rschema)
 	self->calculate_size = avro_resolved_writer_calculate_size_;
 	self->free_iface = avro_resolved_writer_free_iface;
 	self->reset_wrappers = NULL;
+}
+
+static avro_resolved_writer_t *
+avro_resolved_writer_create(avro_schema_t wschema, avro_schema_t rschema)
+{
+	avro_resolved_writer_t  *self = (avro_resolved_writer_t *) avro_new(avro_resolved_writer_t);
+	memset(self, 0, sizeof(avro_resolved_writer_t));
+	avro_resolved_writer_init_iface(self, wschema, rschema);
 	return self;
 }
 
@@ -1735,18 +1741,57 @@ try_array(memoize_state_t *state, avro_resolved_writer_t **self,
  * enum
  */
 
+typedef struct avro_resolved_enum_writer {
+	avro_resolved_writer_t  parent;
+	int  symbol_count;
+	/* Map writer index to reader index with -1 when no mapping */
+	int  *index_mapping;
+} avro_resolved_enum_writer_t;
+
+static void
+avro_resolved_enum_writer_free_iface(avro_resolved_writer_t *iface,
+				     st_table *freeing)
+{
+	AVRO_UNUSED(freeing);
+	avro_resolved_enum_writer_t  *eiface =
+	    container_of(iface, avro_resolved_enum_writer_t, parent);
+	if (eiface->index_mapping != NULL) {
+		avro_free(eiface->index_mapping,
+			  eiface->symbol_count * sizeof(int));
+	}
+	avro_schema_decref(iface->wschema);
+	avro_schema_decref(iface->rschema);
+	avro_freet(avro_resolved_enum_writer_t, eiface);
+}
+
 static int
 avro_resolved_writer_set_enum(const avro_value_iface_t *viface,
 			      void *vself, int val)
 {
 	int  rval;
 	const avro_resolved_writer_t  *iface =
 	    container_of(viface, avro_resolved_writer_t, parent);
+	const avro_resolved_enum_writer_t  *eiface =
+	    container_of(iface, avro_resolved_enum_writer_t, parent);
 	avro_value_t  *self = (avro_value_t *) vself;
 	avro_value_t  dest;
+
+	if (val < 0 || val >= eiface->symbol_count) {
+		avro_set_error("Invalid enum value %d for %s", val,
+			       avro_schema_name(iface->wschema));
+		return EINVAL;
+	}
+	int  reader_val = eiface->index_mapping[val];
+	if (reader_val < 0) {
+		avro_set_error("Symbol %s isn't in the reader's %s enum",
+			       avro_schema_enum_get(iface->wschema, val),
+			       avro_schema_name(iface->wschema));
+		return EINVAL;
+	}
+
 	check(rval, avro_resolved_writer_get_real_dest(iface, self, &dest));
-	DEBUG("Storing %d into %p", val, dest.self);
-	return avro_value_set_enum(&dest, val);
+	DEBUG("Storing %d (writer %d) into %p", reader_val, val, dest.self);
+	return avro_value_set_enum(&dest, reader_val);
 }
 
 static int
@@ -1764,9 +1809,35 @@ try_enum(memoize_state_t *state, avro_resolved_writer_t **self,
 		const char  *rname = avro_schema_name(rschema);
 
 		if (strcmp(wname, rname) == 0) {
-			*self = avro_resolved_writer_create(wschema, root_rschema);
+			int  symbol_count =
+			    avro_schema_enum_number_of_symbols(wschema);
+			int  *index_mapping =
+			    (int *) avro_calloc(symbol_count, sizeof(int));
+			int  wi;
+
+			for (wi = 0; wi < symbol_count; wi++) {
+				const char  *symbol =
+				    avro_schema_enum_get(wschema, wi);
+				index_mapping[wi] = avro_schema_enum_get_by_name
+				    (rschema, symbol);
+			}
+
+			avro_resolved_enum_writer_t  *eself =
+			    (avro_resolved_enum_writer_t *)
+			    avro_new(avro_resolved_enum_writer_t);
+			memset(eself, 0, sizeof(avro_resolved_enum_writer_t));
+			avro_resolved_writer_init_iface(&eself->parent, wschema,
+							root_rschema);
+
+			eself->parent.free_iface =
+			    avro_resolved_enum_writer_free_iface;
+			eself->parent.parent.set_enum =
+			    avro_resolved_writer_set_enum;
+			eself->symbol_count = symbol_count;
+			eself->index_mapping = index_mapping;
+
+			*self = &eself->parent;
 			avro_memoize_set(&state->mem, wschema, root_rschema, *self);
-			(*self)->parent.set_enum = avro_resolved_writer_set_enum;
 		}
 	}
 	return 0;

lang/c/tests/CMakeLists.txt

@@ -88,3 +88,4 @@ add_avro_test_checkmem(test_avro_1691)
 add_avro_test_checkmem(test_avro_1906)
 add_avro_test_checkmem(test_avro_1904)
 add_avro_test_checkmem(test_avro_4246)
+add_avro_test_checkmem(test_avro_4358)

lang/c/tests/test_avro_4358.c

@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied.  See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "avro.h"
+
+#define try(call, msg) \
+	do { \
+		if (call) { \
+			fprintf(stderr, msg ":\n  %s\n", avro_strerror()); \
+			return EXIT_FAILURE; \
+		} \
+	} while (0)
+
+static const char  WRITER_JSON[] =
+"{ \"type\": \"enum\", \"name\": \"Color\","
+"  \"symbols\": [\"RED\", \"GREEN\", \"BLUE\"] }";
+
+/* Reorders the symbols and leaves out GREEN */
+static const char  READER_JSON[] =
+"{ \"type\": \"enum\", \"name\": \"Color\","
+"  \"symbols\": [\"BLUE\", \"RED\"] }";
+
+int main(void)
+{
+	avro_schema_t  wschema;
+	avro_schema_t  rschema;
+	avro_value_t  src;
+	avro_value_t  dest;
+	avro_value_t  resolved_writer;
+	avro_value_t  resolved_reader;
+	int  val;
+
+	try(avro_schema_from_json_literal(WRITER_JSON, &wschema),
+	    "Cannot parse writer schema");
+	try(avro_schema_from_json_literal(READER_JSON, &rschema),
+	    "Cannot parse reader schema");
+
+	avro_value_iface_t  *wclass = avro_generic_class_from_schema(wschema);
+	avro_value_iface_t  *rclass = avro_generic_class_from_schema(rschema);
+	avro_value_iface_t  *writer_resolver =
+	    avro_resolved_writer_new(wschema, rschema);
+	avro_value_iface_t  *reader_resolver =
+	    avro_resolved_reader_new(wschema, rschema);
+	if (writer_resolver == NULL || reader_resolver == NULL) {
+		fprintf(stderr, "Cannot create resolvers:\n  %s\n",
+			avro_strerror());
+		return EXIT_FAILURE;
+	}
+
+	try(avro_generic_value_new(wclass, &src),
+	    "Cannot create writer value");
+	try(avro_generic_value_new(rclass, &dest),
+	    "Cannot create reader value");
+	try(avro_resolved_writer_new_value(writer_resolver, &resolved_writer),
+	    "Cannot create resolved writer value");
+	try(avro_resolved_reader_new_value(reader_resolver, &resolved_reader),
+	    "Cannot create resolved reader value");
+	avro_resolved_writer_set_dest(&resolved_writer, &dest);
+	avro_resolved_reader_set_source(&resolved_reader, &src);
+
+	/* RED is index 0 for the writer, and index 1 for the reader */
+	try(avro_value_set_enum(&resolved_writer, 0), "Cannot set RED");
+	try(avro_value_get_enum(&dest, &val), "Cannot get enum value");
+	if (val != 1) {
+		fprintf(stderr, "Resolved writer didn't match RED by name\n");
+		return EXIT_FAILURE;
+	}
+
+	try(avro_value_set_enum(&src, 0), "Cannot set RED");
+	try(avro_value_get_enum(&resolved_reader, &val),
+	    "Cannot get enum value");
+	if (val != 1) {
+		fprintf(stderr, "Resolved reader didn't match RED by name\n");
+		return EXIT_FAILURE;
+	}
+
+	/* GREEN isn't in the reader's enum */
+	if (avro_value_set_enum(&resolved_writer, 1) == 0) {
+		fprintf(stderr, "Resolved writer accepted a missing symbol\n");
+		return EXIT_FAILURE;
+	}
+
+	try(avro_value_set_enum(&src, 1), "Cannot set GREEN");
+	if (avro_value_get_enum(&resolved_reader, &val) == 0) {
+		fprintf(stderr, "Resolved reader accepted a missing symbol\n");
+		return EXIT_FAILURE;
+	}
+
+	avro_value_decref(&resolved_reader);
+	avro_value_decref(&resolved_writer);
+	avro_value_decref(&dest);
+	avro_value_decref(&src);
+	avro_value_iface_decref(reader_resolver);
+	avro_value_iface_decref(writer_resolver);
+	avro_value_iface_decref(rclass);
+	avro_value_iface_decref(wclass);
+	avro_schema_decref(rschema);
+	avro_schema_decref(wschema);
+	return EXIT_SUCCESS;
+}