DM-25826: lsst.alert.packet reader should iterate over alerts

python/lsst/alert/packet/bin/validateAvroRoundTrip.py

@@ -35,12 +35,14 @@
 
 import lsst.alert.packet
 
-# The default filename of the root of the alert schema.
-SCHEMA_FILENAME = "lsst.alert.avsc"
-
 # The default filename of per-schema sample alert data.
 SAMPLE_FILENAME = "alert.json"
 
+
+def schema_filename(major_version, minor_version):
+    return f"lsst.v{major_version}_{minor_version}.alert.avsc"
+
+
 def check_file_round_trip(baseline, received_data):
     """Assert that the contents of baseline is equal to received_data.
 
@@ -81,8 +83,10 @@ def main():
         schema_major, schema_minor = args.schema_version.split(".")
     schema_root = lsst.alert.packet.get_schema_path(schema_major, schema_minor)
 
-    alert_schema = lsst.alert.packet.Schema.from_file(os.path.join(schema_root,
-                                                            SCHEMA_FILENAME))
+    alert_schema = lsst.alert.packet.Schema.from_file(
+        os.path.join(schema_root,
+                     schema_filename(schema_major, schema_minor)),
+    )
     if args.input_data:
         input_data = args.input_data
     else:

python/lsst/alert/packet/io.py

@@ -21,7 +21,7 @@
 
 """Routines for loading data from files.
 """
-
+import itertools
 import os.path
 
 import fastavro
@@ -68,5 +68,21 @@ def retrieve_alerts(fp, reader_schema=None):
     except Exception as e:
         raise RuntimeError(f"failed to find alert data in "
                            f"{fp.name if hasattr(fp, 'name') else 'stream'}") from e
-    records = [rec for rec in reader]
-    return Schema(reader.writer_schema), records
+
+    # Peek at one record so that reader.writer_schema is populated, since it
+    # gets loaded lazily. If you don't do this, then reader.writer_schema would
+    # be None, which means Schema(reader.writer_schema) would get an empty
+    # value.
+    #
+    # It would be simpler to do something like 'records = list(reader)', but
+    # that would require loading all the records into memory in one gulp. Since
+    # alert files can be huge, even terabytes, the extra complexity here is
+    # worth it.
+    try:
+        first_record = next(reader)
+        records = itertools.chain([first_record], reader)
+    except StopIteration:
+        # The file has zero records in it. It might still have a schema, though.
+        records = []
+    writer_schema = Schema(reader.writer_schema)
+    return writer_schema, records

python/lsst/alert/packet/schema.py

@@ -172,7 +172,6 @@ class Schema(object):
     """
     def __init__(self, schema_definition):
         self.definition = resolve_schema_definition(schema_definition)
-        fastavro.schema._schema.SCHEMA_DEFS.clear()
 
     def serialize(self, record):
         """Create an Avro representation of data following this schema.
@@ -220,19 +219,9 @@ def validate(self, record):
         -------
         valid : `bool`
             Whether or not the data complies with the schema.
-
-        Notes
-        -----
-        Validating the against the schema requires that the fastavro cache
-        (``SCHEMA_DEFS``) be populated, but that can only be the case for one
-        version of the schema at once. Hence we populate, check for validity,
-        and flush the cache.
         """
         fastavro.parse_schema(self.definition)
-        try:
-            return fastavro.validate(record, self.definition)
-        finally:
-            fastavro.schema._schema.SCHEMA_DEFS.clear()
+        return fastavro.validate(record, self.definition)
 
     def store_alerts(self, fp, records):
         """Store alert packets to the given I/O stream.
@@ -267,7 +256,8 @@ def retrieve_alerts(self, fp):
             Alert records.
         """
         from .io import retrieve_alerts
-        return retrieve_alerts(fp, reader_schema=self)
+        schema, records = retrieve_alerts(fp, reader_schema=self)
+        return schema, records
 
     def __eq__(self, other):
         """Compare schemas for equality.
@@ -278,7 +268,7 @@ def __eq__(self, other):
         return self.definition == other.definition
 
     @classmethod
-    def from_file(cls, filename=None, root_name="lsst.alert"):
+    def from_file(cls, filename=None, root_name="lsst.v3_0.alert"):
         """Instantiate a `Schema` by reading its definition from the filesystem.
 
         Parameters