AVRO-3760: [python] Fix resolution of future enum with default symbol

lang/py/avro/io.py

@@ -766,6 +766,9 @@ def read_enum(self, writers_schema: avro.schema.EnumSchema, readers_schema: avro
         # read data
         index_of_symbol = decoder.read_int()
         if index_of_symbol >= len(writers_schema.symbols):
+            default = writers_schema.default
+            if default is not None:
+                return default
             raise avro.errors.SchemaResolutionException(
                 f"Can't access enum index {index_of_symbol} for enum with {len(writers_schema.symbols)} symbols", writers_schema, readers_schema
             )

lang/py/avro/schema.py

@@ -715,6 +715,15 @@ def symbols(self) -> Sequence[str]:
             return symbols
         raise Exception
 
+    @property
+    def default(self) -> Union[str, None]:
+        symbol = self.get_prop("default")
+        if isinstance(symbol, str):
+            return symbol
+        if symbol is None:
+            return None
+        raise avro.errors.InvalidDefault(f"Enum default '{symbol}' is not a valid member of symbols '{self.symbols}'")
+
     @property
     def doc(self):
         return self.get_prop("doc")

lang/py/avro/test/test_io.py

@@ -714,6 +714,116 @@ def test_type_exception_record(self) -> None:
         ):
             write_datum(datum_to_write, writers_schema)
 
+    def test_can_read_future_enum_symbol_with_default(self) -> None:
+        default_symbol = "unknown"
+        future_symbol = "crc32_be"
+
+        readers_schema = avro.schema.parse(
+            json.dumps(
+                {
+                    "fields": [
+                        {
+                            "name": "checksum_algorithm",
+                            "type": {
+                                "name": "ChecksumAlgorithm",
+                                "symbols": [default_symbol, "xxhash3_64_be"],
+                                "type": "enum",
+                                "default": default_symbol,
+                            },
+                        },
+                    ],
+                    "name": "Test",
+                    "type": "record",
+                }
+            )
+        )
+        # Writer adds the "crc32_be" symbol.
+        writers_schema = avro.schema.parse(
+            json.dumps(
+                {
+                    "fields": [
+                        {
+                            "name": "checksum_algorithm",
+                            "type": {
+                                "name": "ChecksumAlgorithm",
+                                "symbols": [
+                                    "unknown",
+                                    "xxhash3_64_be",
+                                    future_symbol,
+                                ],
+                                "type": "enum",
+                                "default": default_symbol,
+                            },
+                        }
+                    ],
+                    "name": "Test",
+                    "type": "record",
+                }
+            )
+        )
+
+        datum_to_write = {"checksum_algorithm": future_symbol}
+
+        buffer, encoder, datum_writer = write_datum(datum_to_write, writers_schema)
+        buffer.seek(0)
+        decoder = avro.io.BinaryDecoder(buffer)
+        reader = avro.io.DatumReader(readers_schema)
+        datum_read = reader.read(decoder)
+        self.assertEqual(datum_read, {"checksum_algorithm": default_symbol})
+
+    def test_raises_error_for_future_enum_symbol_without_default(self) -> None:
+        future_symbol = "crc32_be"
+
+        readers_schema = avro.schema.parse(
+            json.dumps(
+                {
+                    "fields": [
+                        {
+                            "name": "checksum_algorithm",
+                            "type": {
+                                "name": "ChecksumAlgorithm",
+                                "symbols": ["xxhash3_64_be"],
+                                "type": "enum",
+                            },
+                        },
+                    ],
+                    "name": "Test",
+                    "type": "record",
+                }
+            )
+        )
+        # Writer adds the "crc32_be" symbol.
+        writers_schema = avro.schema.parse(
+            json.dumps(
+                {
+                    "fields": [
+                        {
+                            "name": "checksum_algorithm",
+                            "type": {
+                                "name": "ChecksumAlgorithm",
+                                "symbols": [
+                                    "xxhash3_64_be",
+                                    future_symbol,
+                                ],
+                                "type": "enum",
+                            },
+                        }
+                    ],
+                    "name": "Test",
+                    "type": "record",
+                }
+            )
+        )
+
+        datum_to_write = {"checksum_algorithm": future_symbol}
+
+        buffer, encoder, datum_writer = write_datum(datum_to_write, writers_schema)
+        buffer.seek(0)
+        decoder = avro.io.BinaryDecoder(buffer)
+        reader = avro.io.DatumReader(readers_schema)
+        with self.assertRaises(avro.errors.SchemaResolutionException):
+            reader.read(decoder)
+
 
 def load_tests(loader: unittest.TestLoader, default_tests: None, pattern: None) -> unittest.TestSuite:
     """Generate test cases across many test schema."""