Log EventDataIds through the SND event and attribute data ETL steps

snd/src/org/labkey/snd/SNDManager.java

@@ -91,6 +91,7 @@
 import org.labkey.snd.security.SNDSecurityManager;
 import org.labkey.snd.trigger.SNDTriggerManager;
 
+import java.nio.ByteBuffer;
 import java.sql.SQLException;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -104,7 +105,9 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.LongSummaryStatistics;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
@@ -153,6 +156,14 @@ public static UserSchema getSndUserSchemaAdminRole(Container c, User u)
 
     public static int MAX_MERGE_ROWS = 2000;
 
+    /** Below the ETL batch size so that the full batches of an initial load suppress the id lists while the smaller batches of an incremental run keep them. */
+    public static final int MAX_LOGGED_IDS = 2000;
+    private static final int LOGGED_IDS_PER_LINE = 250;
+    private static final int LOGGED_PAIRS_PER_LINE = 50;
+
+    /** The incremental filter column of both SND ETL source views. */
+    public static final String SOURCE_ROWVERSION_COLUMN = "timestamp";
+
     public static Logger getLogger(Map<Enum, Object> configParameters, Class<?> clazz)
     {
         Logger log = null;
@@ -164,6 +175,87 @@ public static Logger getLogger(Map<Enum, Object> configParameters, Class<?> claz
         return log;
     }
 
+    /**
+     * Writes an id set to the ETL job log so that the id sets logged by different ETL steps of the same run can be
+     * diffed against each other. Chunked because a single line of thousands of ids is unreadable, and capped because
+     * the initial full data load would otherwise write the entire table to the log.
+     */
+    public static void logIds(Logger log, String message, Collection<Integer> ids)
+    {
+        if (!log.isDebugEnabled())
+            return;
+
+        log.debug(message + " Count: " + ids.size() + ".");
+
+        if (ids.isEmpty())
+            return;
+
+        if (ids.size() > MAX_LOGGED_IDS)
+        {
+            log.debug("Id list omitted, more than " + MAX_LOGGED_IDS + " ids.");
+            return;
+        }
+
+        List<Integer> sorted = ids.stream().filter(Objects::nonNull).sorted().collect(Collectors.toList());
+        for (List<Integer> chunk : ListUtils.partition(sorted, LOGGED_IDS_PER_LINE))
+            log.debug("    " + StringUtils.join(chunk, ", "));
+    }
+
+    /**
+     * SQL Server hands a rowversion back as binary(8); the ETL's own persisted window state returns it as a number.
+     * Read big-endian, matching how the incremental filter logs its bounds, so the two can be compared directly.
+     */
+    @Nullable
+    public static Long toRowversion(@Nullable Object o)
+    {
+        if (o instanceof byte[] bytes && 8 == bytes.length)
+            return ByteBuffer.wrap(bytes).getLong();
+        if (o instanceof Number n)
+            return n.longValue();
+        return null;
+    }
+
+    /**
+     * Logs the rowversion span of a batch so it can be placed against the incremental window the ETL logged for the
+     * run. Both SND source views draw their rowversions from the same source database, so the spans the two steps
+     * report are on one sequence and comparable.
+     */
+    public static void logRowversionRange(Logger log, String message, Collection<Map<String, Object>> rows)
+    {
+        if (!log.isDebugEnabled())
+            return;
+
+        LongSummaryStatistics stats = rows.stream()
+                .map(row -> toRowversion(row.get(SOURCE_ROWVERSION_COLUMN)))
+                .filter(Objects::nonNull)
+                .mapToLong(Long::longValue)
+                .summaryStatistics();
+
+        if (0 == stats.getCount())
+            log.debug(message + " No source rowversions in this batch.");
+        else
+            log.debug(message + " Rowversions " + stats.getMin() + " to " + stats.getMax() + " over " + stats.getCount() + " rows.");
+    }
+
+    /**
+     * Pairs each id with its source rowversion. Logged separately from the bare list the same set gets from logIds,
+     * which stays free of annotations so it can be diffed against the other step's list.
+     */
+    public static void logIdRowversions(Logger log, String message, Collection<Integer> ids, Map<Integer, Long> rowversions)
+    {
+        if (!log.isDebugEnabled() || ids.isEmpty() || ids.size() > MAX_LOGGED_IDS)
+            return;
+
+        log.debug(message);
+
+        List<String> pairs = ids.stream().filter(Objects::nonNull).sorted()
+                .map(id -> id + ":" + rowversions.get(id))
+                .collect(Collectors.toList());
+
+        for (List<String> chunk : ListUtils.partition(pairs, LOGGED_PAIRS_PER_LINE))
+            log.debug("    " + StringUtils.join(chunk, ", "));
+    }
+
     public static String getPackageName(int id)
     {
         return PackageDomainKind.getPackageKindName() + "-" + id;

snd/src/org/labkey/snd/query/AttributeDataTable.java

@@ -149,6 +149,10 @@ public QueryUpdateService getUpdateService()
 
     protected class UpdateService extends SNDQueryUpdateService
     {
+        /** Bounds the source ordering check below. It costs one retained URI per distinct EventDataId, and an ungrouped source would otherwise log once per row. */
+        private static final int MAX_TRACKED_URIS = 50_000;
+        private static final int MAX_ORDER_WARNINGS = 10;
+
         private final SNDManager _sndManager = SNDManager.get();
         private final SNDService _sndService = SNDService.get();
         private final DbSchema _expSchema = OntologyManager.getExpSchema();
@@ -232,6 +236,26 @@ private List<Map<String, Object>> updateObjectProperty(User user, Container cont
         {
             logger.info("Begin updating exp.ObjectProperty.");
 
+            // An EventDataId gets one source row per attribute; keep the newest, since that is the one that pulled it into the window.
+            boolean trackRowversions = logger.isDebugEnabled();
+            Set<Integer> incomingEventDataIds = new HashSet<>();
+            Map<Integer, Long> rowversionByEventDataId = new HashMap<>();
+            for (Map<String, Object> row : data)
+            {
+                Integer eventDataId = (Integer) row.get("EventDataId");
+                incomingEventDataIds.add(eventDataId);
+
+                if (trackRowversions)
+                {
+                    Long rowversion = SNDManager.toRowversion(row.get(SNDManager.SOURCE_ROWVERSION_COLUMN));
+                    if (null != rowversion)
+                        rowversionByEventDataId.merge(eventDataId, rowversion, Math::max);
+                }
+            }
+
+            SNDManager.logIds(logger, "Source rows: " + data.size() + ". EventDataIds in this batch:", incomingEventDataIds);
+            SNDManager.logRowversionRange(logger, "Source span of this batch.", data);
+
             int inserted = 0;
 
             String prevUri = null;
@@ -240,6 +264,10 @@ private List<Map<String, Object>> updateObjectProperty(User user, Container cont
             boolean found = false;
 
             Set<Integer> cacheEventIds = new HashSet<>();
+            Set<Integer> writtenEventDataIds = new HashSet<>();
+            Set<String> flushedUris = new HashSet<>();
+            boolean checkOrdering = logger.isDebugEnabled();
+            int outOfOrderFlushes = 0;
 
             for(Map<String, Object> row : data)
             {
@@ -255,7 +283,8 @@ private List<Map<String, Object>> updateObjectProperty(User user, Container cont
                 //add to list of cached narrative rows to delete
                 cacheEventIds.add((Integer) row.get("EventId"));
 
-                String objectURI = getObjectURI((Integer) row.get("EventDataId"), container);
+                Integer eventDataId = (Integer) row.get("EventDataId");
+                String objectURI = getObjectURI(eventDataId, container);
                 if (prevUri == null)
                     prevUri = objectURI;
 
@@ -292,11 +321,11 @@ else if (stringValue != null)
                             {
                                 if (pd.getLookupSchema() != null && pd.getLookupQuery() != null)
                                 {
-                                    logger.info("Value null for property " + pd.getName() + ". Value skipped. Verify lookup " + pd.getLookupSchema() + "." + pd.getLookupQuery() + " contains " + stringValue);
+                                    logger.info("Value null for property " + pd.getName() + ", EventDataId: " + eventDataId + ". Value skipped. Verify lookup " + pd.getLookupSchema() + "." + pd.getLookupQuery() + " contains " + stringValue);
                                 }
                                 else
                                 {
-                                    logger.info("Value null for property " + pd.getName() + ". Value skipped.");
+                                    logger.info("Value null for property " + pd.getName() + ", EventDataId: " + eventDataId + ". Value skipped.");
                                 }
                             }
 
@@ -320,10 +349,27 @@ else if (stringValue != null)
                                     if (!prevUri.equals(objectURI))
                                     {
                                         inserted = insertObject(container, user, prevUri, prevObjProps, pkgId, inserted, logger);
+
+                                        // Properties are only flushed when the URI changes, so a URI seen twice means the source
+                                        // did not arrive grouped by EventDataId and the ORDER BY in v_snd_attributeData was lost.
+                                        if (checkOrdering)
+                                        {
+                                            if (!flushedUris.add(prevUri) && ++outOfOrderFlushes <= MAX_ORDER_WARNINGS)
+                                                logger.debug("Source rows are not grouped by EventDataId; exp.ObjectProperty for {} was written in more than one pass.", prevUri);
+
+                                            if (flushedUris.size() >= MAX_TRACKED_URIS)
+                                            {
+                                                logger.debug("More than {} EventDataIds in this batch; ending the source ordering check.", MAX_TRACKED_URIS);
+                                                flushedUris.clear();
+                                                checkOrdering = false;
+                                            }
+                                        }
+
                                         prevUri = objectURI;
                                         prevObjProps = new ArrayList<>();
                                     }
                                     prevObjProps.add(oprop);
+                                    writtenEventDataIds.add(eventDataId);
                                 }
                             }
 
@@ -332,12 +378,16 @@ else if (stringValue != null)
                     }
                     if (!found)
                     {
-                        throw new RuntimeException("Attribute metadata not found for key: '" + key + "' in package: " + pkgId);
+                        throw new RuntimeException("Attribute metadata not found for key: '" + key + "' in package: " + pkgId
+                                + ", EventDataId: " + eventDataId + ". Aborting, leaving all " + incomingEventDataIds.size()
+                                + " EventDataIds in this batch with the attribute values the _SND Event Data step already cleared.");
                     }
                 }
                 else
                 {
-                    throw new RuntimeException("Package metadata not found for package id: " + pkgId);
+                    throw new RuntimeException("Package metadata not found for package id: " + pkgId
+                            + ", EventDataId: " + eventDataId + ". Aborting, leaving all " + incomingEventDataIds.size()
+                            + " EventDataIds in this batch with the attribute values the _SND Event Data step already cleared.");
                 }
             }
 
@@ -347,8 +397,28 @@ else if (stringValue != null)
             }
 
             OntologyManager.clearPropertyCache();
+
             logger.info("End updating exp.ObjectProperty. Inserted/Updated " + inserted + " rows.");
 
+            SNDManager.logIds(logger, "EventDataIds written:", writtenEventDataIds);
+
+            if (outOfOrderFlushes > MAX_ORDER_WARNINGS)
+                logger.debug("{} objectURIs in total were written in more than one pass; further messages were suppressed.", outOfOrderFlushes);
+
+            // Collect only the misses; copying the incoming set would double its footprint on a full load.
+            Set<Integer> unwritten = new HashSet<>();
+            for (Integer id : incomingEventDataIds)
+            {
+                if (!writtenEventDataIds.contains(id))
+                    unwritten.add(id);
+            }
+
+            if (!unwritten.isEmpty())
+            {
+                SNDManager.logIds(logger, "EventDataIds present in the source rows but left with no attribute values written:", unwritten);
+                SNDManager.logIdRowversions(logger, "Rowversions of those EventDataIds, to place them against the incremental window of this run:", unwritten, rowversionByEventDataId);
+            }
+
             _sndManager.updateNarrativeCache(container, user, cacheEventIds, logger);
 
             return data;