Skip to content

Log EventDataIds through the SND event and attribute data ETL steps #985

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
labkey-martyp wants to merge 4 commits into
base: release26.3-SNAPSHOT
Choose a base branch
from
Open

Log EventDataIds through the SND event and attribute data ETL steps #985

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5e4c282
Log EventDataIds through the SND event and attribute data ETL steps
labkey-martyp Aug 18, 2026
23c45e3
Keep the SND attribute-clearing diagnostic off the merge critical path
labkey-martyp Aug 19, 2026
74267b5
Name the EventDataId in the skipped-attribute-value log lines
labkey-martyp Aug 19, 2026
dbd6710
Log EventDataIds on the non-merge paths and drop diagnostics to debug
labkey-martyp Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions snd/src/org/labkey/snd/SNDManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
Expand Down Expand Up @@ -153,6 +154,10 @@ 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;

public static Logger getLogger(Map<Enum, Object> configParameters, Class<?> clazz)
{
Logger log = null;
Expand All @@ -164,6 +169,32 @@ 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, ", "));
}

public static String getPackageName(int id)
{
return PackageDomainKind.getPackageKindName() + "-" + id;
Expand Down
63 changes: 58 additions & 5 deletions snd/src/org/labkey/snd/query/AttributeDataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -232,6 +236,12 @@ private List<Map<String, Object>> updateObjectProperty(User user, Container cont
{
logger.info("Begin updating exp.ObjectProperty.");

Set<Integer> incomingEventDataIds = new HashSet<>();
for (Map<String, Object> row : data)
incomingEventDataIds.add((Integer) row.get("EventDataId"));

SNDManager.logIds(logger, "Source rows: " + data.size() + ". EventDataIds in this batch:", incomingEventDataIds);

int inserted = 0;

String prevUri = null;
Expand All @@ -240,6 +250,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)
{
Expand All @@ -255,7 +269,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;

Expand Down Expand Up @@ -292,11 +307,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.");
}
}

Expand All @@ -320,10 +335,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);
}
}

Expand All @@ -332,12 +364,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.");
}
}

Expand All @@ -347,8 +383,25 @@ 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.updateNarrativeCache(container, user, cacheEventIds, logger);

return data;
Expand Down
99 changes: 94 additions & 5 deletions snd/src/org/labkey/snd/query/EventDataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.labkey.api.data.JdbcType;
import org.labkey.api.data.SQLFragment;
import org.labkey.api.data.SqlExecutor;
import org.labkey.api.data.SqlSelector;
import org.labkey.api.data.TableInfo;
import org.labkey.api.dataiterator.DataIteratorBuilder;
import org.labkey.api.dataiterator.DataIteratorContext;
Expand All @@ -48,7 +49,9 @@

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -108,6 +111,9 @@ public QueryUpdateService getUpdateService()

protected static class UpdateService extends SNDQueryUpdateService
{
/** Keeps the ObjectURI IN clause well under the SQL Server parameter limit. */
private static final int URI_CHUNK_SIZE = 500;

private final SNDManager _sndManager = SNDManager.get();
private final SNDService _sndService = SNDService.get();
private final DbSchema _expSchema = OntologyManager.getExpSchema();
Expand All @@ -122,6 +128,62 @@ private String getObjectURI(Integer eventDataId, Container c)
return _sndManager.generateLsid(c, String.valueOf(eventDataId));
}

/**
* EventDataIds in this batch whose exp.Object currently carries attribute values. Deleting the exp.Object row
* cascades to exp.ObjectProperty, so these are the values the merge destroys; only the _SND Attribute Data ETL
* step re-inserts them, and it computes its incremental window independently of this step's.
*/
private Set<Integer> getEventDataIdsWithAttributeData(Container container, Map<String, Integer> eventDataIdsByUri)
{
Set<Integer> withAttributeData = new HashSet<>();
List<String> uris = new ArrayList<>(eventDataIdsByUri.keySet());

for (int i = 0; i < uris.size(); i += URI_CHUNK_SIZE)
{
List<String> chunk = uris.subList(i, Math.min(i + URI_CHUNK_SIZE, uris.size()));

// EXISTS rather than a join: both indexes (UQ_Object on ObjectURI, PK_ObjectProperty on ObjectId)
// are seeks, and the semi-join stops at the first property instead of reading all of them per object.
SQLFragment sql = new SQLFragment("SELECT o.ObjectURI FROM ")
.append(OntologyManager.getTinfoObject(), "o")
.append(" WHERE o.Container = ?").add(container.getId())
.append(" AND EXISTS (SELECT 1 FROM ").append(OntologyManager.getTinfoObjectProperty(), "op")
.append(" WHERE op.ObjectId = o.ObjectId)")
.append(" AND o.ObjectURI").appendInClause(chunk, _expSchema.getSqlDialect());

new SqlSelector(_expSchema, sql).getCollection(String.class)
.forEach(uri -> withAttributeData.add(eventDataIdsByUri.get(uri)));
}

return withAttributeData;
}

/**
* Diagnostic only, so a failure here must not abort the merge. Skipped above the cap logIds lists at, where the
* chunked queries would cost hundreds of round trips to produce a bare count.
*/
private void logAttributeDataToBeCleared(Container container, Map<String, Integer> eventDataIdsByUri, Logger log)
{
if (!log.isDebugEnabled())
return;

if (eventDataIdsByUri.size() > SNDManager.MAX_LOGGED_IDS)
{
log.debug("More than " + SNDManager.MAX_LOGGED_IDS + " EventDataIds in this batch; skipping the check for attribute values about to be cleared.");
return;
}

try
{
SNDManager.logIds(log, "Attribute values about to be cleared by this merge; the _SND Attribute Data step must re-insert them.",
getEventDataIdsWithAttributeData(container, eventDataIdsByUri));
}
catch (Exception e)
{
log.debug("Could not determine which EventDataIds have attribute values; continuing with the merge.", e);
}
}

@Override
public int mergeRows(User user, Container container, DataIteratorBuilder rows, BatchValidationException errors,
@Nullable Map<Enum, Object> configParameters, Map<String, Object> extraScriptContext)
Expand Down Expand Up @@ -158,14 +220,27 @@ public int mergeRows(User user, Container container, DataIteratorBuilder rows, B
log.info("Merging rows.");

log.info("Begin updating exp.Object table.");
int count = 0;
for(Map<String, Object> map : data)

Map<String, Integer> eventDataIdsByUri = new LinkedHashMap<>();
for (Map<String, Object> map : data)
{
String objectURI = getObjectURI((Integer) map.get("EventDataId"), container);
Integer eventDataId = (Integer) map.get("EventDataId");
String objectURI = getObjectURI(eventDataId, container);

//update snd.EventData row with objectURI
map.put("ObjectURI", objectURI);

eventDataIdsByUri.put(objectURI, eventDataId);
}

SNDManager.logIds(log, "EventDataIds merged into snd.EventData by this batch:", eventDataIdsByUri.values());
logAttributeDataToBeCleared(container, eventDataIdsByUri, log);

int count = 0;
for(Map<String, Object> map : data)
{
String objectURI = (String) map.get("ObjectURI");

//delete row from exp.Object
OntologyManager.deleteOntologyObjects(container, objectURI);

Expand Down Expand Up @@ -215,9 +290,11 @@ public int importRows(User user, Container container, DataIteratorBuilder rows,

log.info("Begin inserting into exp.Object.");
int count = 0;
Set<Integer> eventDataIds = new HashSet<>();
for(Map<String, Object> map : data)
{
String objectURI = getObjectURI((Integer) map.get("EventDataId"), container);
Integer eventDataId = (Integer) map.get("EventDataId");
String objectURI = getObjectURI(eventDataId, container);

//update snd.EventData row with objectURI
map.put("ObjectURI", objectURI);
Expand All @@ -228,13 +305,18 @@ public int importRows(User user, Container container, DataIteratorBuilder rows,
//add to list of cached narrative rows to delete
cacheData.add((Integer) map.get("EventId"));

eventDataIds.add(eventDataId);

count++;
//TODO: Count in exp.Object is not going to be the same as in snd.EventData - need to figure out how to get the count to log
if(count % 1000 == 0)
log.info("Inserted " + count + " rows in exp.Object table.");
}
log.info("End inserting into exp.Object. Inserted total of " + count + " rows.");

// These rows get a fresh exp.Object with no properties, so they depend on the _SND Attribute Data step just as much as the merged ones do.
SNDManager.logIds(log, "EventDataIds inserted into snd.EventData by this batch:", eventDataIds);

DataIteratorBuilder rowsWithObjectURI = new ListofMapsDataIterator.Builder(data.get(0).keySet(), data);

_sndManager.updateNarrativeCache(container, user, cacheData, log);
Expand Down Expand Up @@ -334,11 +416,15 @@ private void deleteFromExpTables(List<Map<String, Object>> oldRows, Container co
{
log.info("Begin deleting from exp.ObjectProperty and exp.Object.");
int count = 0;
Set<Integer> eventDataIds = new HashSet<>();

//This will be a cascading delete across exp.ObjectProperty, exp.Object, and snd.EventData
for (Map<String, Object> map : oldRows)
{
String objectURI = getObjectURI((Integer) map.get("EventDataId"), container);
Integer eventDataId = (Integer) map.get("EventDataId");
String objectURI = getObjectURI(eventDataId, container);

eventDataIds.add(eventDataId);
OntologyObject obj = OntologyManager.getOntologyObject(container, objectURI);

//delete row from exp.ObjectProperty
Expand All @@ -355,6 +441,9 @@ private void deleteFromExpTables(List<Map<String, Object>> oldRows, Container co
}

log.info("End deleting from exp.ObjectProperty and exp.Object. Deleted total of " + count + " rows.");

// Without these the deleted rows read as attribute data the _SND Attribute Data step failed to write.
SNDManager.logIds(log, "EventDataIds deleted from snd.EventData by this batch:", eventDataIds);
}

private int deleteAllFromExpTables(Logger log)
Expand Down