Merge pull request #286 from NASA-PDS/ldd_write_wait

Make harvest wait for LDD to be ready when it is updated.

Author: tloubrieu-jpl

src/main/java/gov/nasa/pds/registry/common/Request.java

@@ -41,6 +41,7 @@ public interface Mapping { // _mapping
   }
   public interface MGet extends Get { // _mget
     public MGet setIds (Collection<String> ids);
+    public MGet setRefresh (boolean refresh);
   }
   public interface Search { // _search
     public Search all(String sortField, int size, String searchAfter);
@@ -52,6 +53,7 @@ public interface Search { // _search
     public Search buildLidvidsFromTermQuery (String fieldname, String value);
     public Search buildListFields(String dataType);
     public Search buildListLdds (String namespace);
+    public Search buildListLddsNoCache (String namespace);
     public Search buildTermQuery (String fieldname, String value);
     public Search buildTermQueryWithoutTermQuery (String yesFieldname, String yesValue, String noFieldname, String noValue);
     public Search buildTheseIds(Collection<String> lids);

src/main/java/gov/nasa/pds/registry/common/connection/aws/MGetImpl.java

@@ -44,4 +44,9 @@ public MGet setIds(Collection<String> ids) {
     this.craftsman.ids(new ArrayList<String>(ids));
     return this;
   }
+  @Override
+  public MGet setRefresh(boolean refresh) {
+    this.craftsman.refresh(refresh);
+    return this;
+  }
 }

src/main/java/gov/nasa/pds/registry/common/connection/aws/SearchImpl.java

@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+
 import org.opensearch.client.opensearch._types.FieldSort;
 import org.opensearch.client.opensearch._types.FieldValue;
 import org.opensearch.client.opensearch._types.SortOptions;
@@ -26,6 +27,7 @@
 
 class SearchImpl implements Search {
   final SearchRequest.Builder craftsman = new SearchRequest.Builder();
+
   private void buildIds (Collection<String> lids, boolean alt) {
     SourceConfig.Builder journeyman = new SourceConfig.Builder();
     if (alt) {
@@ -88,6 +90,20 @@ public Search buildListLdds(String namespace) {
     this.craftsman.source(new SourceConfig.Builder().filter(new SourceFilter.Builder().includes("date", "attr_name").build()).build());
     return this;
   }
+
+  @Override
+  public Search buildListLddsNoCache(String namespace) {
+    BoolQuery.Builder journeyman = new BoolQuery.Builder()
+        .must(this.matchQuery("class_ns", "registry").build(),
+            this.matchQuery("class_name", "LDD_Info").build(),
+            this.matchQuery("attr_ns", namespace).build());
+    this.craftsman.query(new Query.Builder().bool(journeyman.build()).build());
+    this.craftsman.requestCache(false);
+    this.craftsman.size(1000); // have no idea why hardcoded but it is (.es.JsonHelper:265
+    this.craftsman.source(new SourceConfig.Builder().filter(new SourceFilter.Builder().includes("date", "attr_name").build()).build());
+    return this;
+  }
+
   @Override
   public Search buildTheseIds(Collection<String> lids) {
     this.buildIds(lids, false);

src/main/java/gov/nasa/pds/registry/common/connection/es/GetImpl.java

@@ -43,6 +43,11 @@ public MGet setIds(Collection<String> ids) {
     return this;
   }
   @Override
+  public MGet setRefresh(boolean refresh) {
+    // no-op: SDK1 ES client does not use this path for AOSS eventual consistency
+    return this;
+  }
+  @Override
   public Get setIndex(String index) {
     this.setIndex(index);
     return this;

src/main/java/gov/nasa/pds/registry/common/connection/es/SearchImpl.java

@@ -28,6 +28,14 @@ public Search buildListLdds(String namespace) {
     this.json = JsonHelper.buildListLddsRequest(namespace);
     return this;
   }
+
+  @Override
+  public Search buildListLddsNoCache(String namespace) {
+    // not needed since the cache is now disabled in the data dictionary index, but we keep the method for backward compatibility
+    this.json = JsonHelper.buildListLddsRequest(namespace);
+    return this;
+  }
+
   @Override
   public Search buildTheseIds(Collection<String> lids) {
     this.json = JsonHelper.buildSearchIdsRequest(lids, lids.size(), true);

src/main/java/gov/nasa/pds/registry/common/es/dao/DataLoader.java

@@ -268,6 +268,7 @@ private int emptyQueue(LinkedHashMap<String, String> todo, Set<String> errorLidv
         bulk.add(item.getKey(), item.getValue());
       }
       Response.Bulk response = this.conFactory.createRestClient().performRequest(bulk);
+            
       failed += processErrors(response, errorLidvids, todo, retry);
       retry++;
 

src/main/java/gov/nasa/pds/registry/common/es/dao/dd/DataDictionaryDao.java

@@ -11,7 +11,7 @@
 import gov.nasa.pds.registry.common.util.Tuple;
 
 
-/**
+  /**
  * Data dictionary DAO (Data Access Object). This class provides methods to read and update data
  * dictionary.
  * 
@@ -52,6 +52,24 @@ public LddVersions getLddInfo(String namespace)
     return client.performRequest(req).lddInfo();
   }
 
+    /**
+   * Get LDD date from data dictionary index in Elasticsearch. Force skip the OpenSearch cache.
+   * 
+   * @param namespace LDD namespace, e.g., "pds", "geom", etc.
+   * @return ISO instant class representing LDD date.
+   * @throws IOException
+   * @throws ResponseException
+   * @throws UnsupportedOperationException
+   * @throws Exception an exception
+   */
+  public LddVersions getLddInfoNoCache(String namespace)
+      throws UnsupportedOperationException, IOException {
+    Request.Search req =
+        client.createSearchRequest().buildListLddsNoCache(namespace).setIndex(indexName + "-dd");
+    return client.performRequest(req).lddInfo();
+  }
+  
+
 
   /**
    * List registered LDDs
@@ -110,5 +128,19 @@ public List<Tuple> getDataTypes(Collection<String> ids)
     return this.client.performRequest(req).dataTypes();
   }
 
-}
+  /**
+   * Same as getDataTypes but forces a shard refresh before the mget. Use only in targeted wait
+   * loops after bulk loading an LDD — do not use in normal query paths.
+   */
+  public List<Tuple> getDataTypesWithRefresh(Collection<String> ids)
+      throws IOException, DataTypeNotFoundException {
+    if (ids == null || ids.isEmpty())
+      return null;
+    Request.MGet mgetReq = client.createMGetRequest();
+    mgetReq.setRefresh(true);
+    Request.Get req = mgetReq.setIds(ids).includeField("es_data_type")
+        .setIndex(this.indexName + "-dd");
+    return this.client.performRequest(req).dataTypes();
+  }
 
+}

src/main/java/gov/nasa/pds/registry/common/es/service/JsonLddLoader.java

@@ -2,6 +2,7 @@
 
 import java.io.File;
 import java.time.Instant;
+import java.util.Collections;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -16,6 +17,7 @@
 import gov.nasa.pds.registry.common.dd.parser.DDAttribute;
 import gov.nasa.pds.registry.common.es.dao.DataLoader;
 import gov.nasa.pds.registry.common.es.dao.dd.DataDictionaryDao;
+import gov.nasa.pds.registry.common.es.dao.dd.DataTypeNotFoundException;
 import gov.nasa.pds.registry.common.es.dao.dd.LddVersions;
 
 
@@ -88,6 +90,10 @@ public void load(File lddFile, String lddFileName, String namespace) throws Exce
       namespace = LddUtils.getLddNamespace(lddFile);
     }
 
+    // get date of the current LDD for this namespace in the registry. If there is no LDD for this namespace, use epoch date.
+    // TODO add the test of overwrite yes/no, remove it from where it is not >?
+
+
     // Get information about LDDs already loaded into the registry (for this namespace)
     LddVersions info = dao.getLddInfo(namespace);
     if (info.files.contains(lddFileName)) {
@@ -118,8 +124,42 @@ public void loadOnly(File lddFile, String lddFileName, String namespace, Instant
     log.info("Creating temporary ES data file " + tempEsDataFile.getAbsolutePath());
 
     try {
-      createEsDataFile(lddFile, lddFileName, namespace, tempEsDataFile, lastDate);
+      String firstFieldId = createEsDataFile(lddFile, lddFileName, namespace, tempEsDataFile, lastDate);
       loader.loadFile(tempEsDataFile);
+
+      // Wait until the LDD_Info sentinel is visible (search with requestCache=false).
+      // This confirms the bulk load completed on the server side.
+      LddVersions info = dao.getLddInfoNoCache(namespace);
+      int nAttempts = 0;
+      int maxAttempts = 30;
+      while (info.isEmpty() && nAttempts < maxAttempts) {
+        Thread.sleep(1000);
+        log.info("Waiting for the new LDD {} to be indexed...", namespace);
+        info = dao.getLddInfoNoCache(namespace);
+      }
+
+      if (info.isEmpty()) {
+        log.warn("The new LDD {} is not indexed after {} seconds. It may be indexed later, but there may be a delay in loading other LDDs for this namespace.", namespace, maxAttempts);
+      } else {
+        log.info("The new LDD {} is indexed with date {}. It may take some time for it to be visible via mget.", namespace, info.lastDate);
+     
+      // On AWS OpenSearch Serverless (AOSS), mget and search use different visibility paths.
+      // Wait until a specific field document is also visible via mget with refresh=true,
+      // so that getDataTypes() calls immediately following this load will succeed.
+      if (firstFieldId != null) {
+        boolean fieldVisible = false;
+        while (!fieldVisible) {
+          try {
+            dao.getDataTypesWithRefresh(Collections.singletonList(firstFieldId));
+            fieldVisible = true;
+          } catch (DataTypeNotFoundException e) {
+            Thread.sleep(1000);
+            log.info("Waiting for field {} of namespace {} to be visible via mget...", firstFieldId, namespace);
+          }
+        }
+      }
+      log.info("Visibility of namespace " + namespace + "has been fully validated.");
+    }
     } finally {
       // Delete temporary file
       tempEsDataFile.delete();
@@ -128,15 +168,30 @@ public void loadOnly(File lddFile, String lddFileName, String namespace, Instant
 
 
   private static class CaaCallback implements ClassAttrAssociationParser.Callback {
-    private LddEsJsonWriter writer;
+    private final LddEsJsonWriter writer;
+    private final String namespace;
+    private final Map<String, DDAttribute> ddAttrCache;
+    private String firstFieldId;
 
-    public CaaCallback(LddEsJsonWriter writer) {
+    public CaaCallback(LddEsJsonWriter writer, String namespace, Map<String, DDAttribute> ddAttrCache) {
       this.writer = writer;
+      this.namespace = namespace;
+      this.ddAttrCache = ddAttrCache;
     }
 
     @Override
     public void onAssociation(String classNs, String className, String attrId) throws Exception {
       writer.writeFieldDefinition(classNs, className, attrId);
+      if (firstFieldId == null && namespace.equals(classNs)) {
+        DDAttribute attr = ddAttrCache.get(attrId);
+        if (attr != null) {
+          firstFieldId = classNs + ":" + className + "/" + attr.attrNs + ":" + attr.attrName;
+        }
+      }
+    }
+
+    public String getFirstFieldId() {
+      return firstFieldId;
     }
   }
 
@@ -149,7 +204,7 @@ public void onAssociation(String classNs, String className, String attrId) throw
    * @param tempEsFile Write to this Elasticsearch file
    * @throws Exception an exception
    */
-  private void createEsDataFile(File lddFile, String lddFileName, String namespace, File tempEsFile,
+  private String createEsDataFile(File lddFile, String lddFileName, String namespace, File tempEsFile,
       Instant lastDate) throws Exception {
     // Parse and cache LDD attributes
     Map<String, DDAttribute> ddAttrCache = new TreeMap<>();
@@ -169,13 +224,15 @@ private void createEsDataFile(File lddFile, String lddFileName, String namespace
       writer.setNamespaceFilter(namespace);
 
       // Parse class attribute associations and write to ES data file
-      CaaCallback ccb = new CaaCallback(writer);
+      CaaCallback ccb = new CaaCallback(writer, namespace, ddAttrCache);
       ClassAttrAssociationParser caaParser = new ClassAttrAssociationParser(lddFile, ccb);
       caaParser.parse();
 
       // Write data dictionary version and date
       writer.writeLddInfo(namespace, lddFileName, attrParser.getImVersion(),
           attrParser.getLddVersion(), attrParser.getLddDate());
+
+      return ccb.getFirstFieldId();
     } finally {
       writer.close();
     }