Add the document db to a list for simpler lookup

6 files changed, 86 insertions, 82 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/application/validation/StreamingValidator.java b/config-model/src/main/java/com/yahoo/vespa/model/application/validation/StreamingValidator.java
index c0ad55fc8f4..0e9d3bf7016 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/application/validation/StreamingValidator.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/application/validation/StreamingValidator.java
@@ -6,12 +6,14 @@ import com.yahoo.document.DataType;
 import com.yahoo.document.NumericDataType;
 import com.yahoo.document.TensorDataType;
 import com.yahoo.documentmodel.NewDocumentReferenceDataType;
+import com.yahoo.schema.Schema;
+import com.yahoo.schema.derived.DerivedConfiguration;
+import com.yahoo.schema.derived.SchemaInfo;
 import com.yahoo.schema.document.Attribute;
 import com.yahoo.schema.document.ImmutableSDField;
 import com.yahoo.schema.document.MatchType;
 import com.yahoo.vespa.model.application.validation.Validation.Context;
 import com.yahoo.vespa.model.search.SearchCluster;
-import com.yahoo.vespa.model.search.StreamingSearchCluster;
 
 import java.util.List;
 import java.util.logging.Level;
@@ -25,20 +27,21 @@ public class StreamingValidator implements Validator {
     public void validate(Context context) {
         List<SearchCluster> searchClusters = context.model().getSearchClusters();
         for (SearchCluster cluster : searchClusters) {
-            if ( ! cluster.isStreaming()) continue;
-
-            var streamingCluster = (StreamingSearchCluster)cluster;
-            warnStreamingAttributes(streamingCluster, context.deployState().getDeployLogger());
-            warnStreamingGramMatching(streamingCluster, context.deployState().getDeployLogger());
-            failStreamingDocumentReferences(context, streamingCluster);
+            for (SchemaInfo schemaInfo : cluster.schemas().values()) {
+                if (schemaInfo.getIndexMode() == SchemaInfo.IndexMode.STREAMING) {
+                    var deployLogger = context.deployState().getDeployLogger();
+                    warnStreamingAttributes(cluster.getClusterName(), schemaInfo.fullSchema(), deployLogger);
+                    warnStreamingGramMatching(cluster.getClusterName(), schemaInfo.fullSchema(), deployLogger);
+                    failStreamingDocumentReferences(cluster.getClusterName(), cluster.getDocumentDB(schemaInfo.name()).getDerivedConfiguration(), context);
+                }
+            }
         }
     }
 
-    private static void warnStreamingGramMatching(StreamingSearchCluster sc, DeployLogger logger) {
-        for (ImmutableSDField sd : sc.derived().getSchema().allConcreteFields()) {
+    private static void warnStreamingGramMatching(String cluster, Schema schema, DeployLogger logger) {
+        for (ImmutableSDField sd : schema.allConcreteFields()) {
             if (sd.getMatching().getType() == MatchType.GRAM) {
-                logger.logApplicationPackage(Level.WARNING, "For streaming search cluster '" +
-                                                            sc.getClusterName() +
+                logger.logApplicationPackage(Level.WARNING, "For search cluster '" + cluster + "', schema '" + schema.getName() +
                                                             "', SD field '" + sd.getName() +
                                                             "': n-gram matching is not supported for streaming search.");
             }
@@ -47,19 +50,16 @@ public class StreamingValidator implements Validator {
 
     /**
      * Warn if one or more attributes are defined in a streaming search cluster SD.
-     *
-     * @param sc     a search cluster to be checked for attributes in streaming search
-     * @param logger a DeployLogger
      */
-    private static void warnStreamingAttributes(StreamingSearchCluster sc, DeployLogger logger) {
-        for (ImmutableSDField sd : sc.derived().getSchema().allConcreteFields()) {
+    private static void warnStreamingAttributes(String cluster, Schema schema, DeployLogger logger) {
+        for (ImmutableSDField sd : schema.allConcreteFields()) {
             if (sd.doesAttributing()) {
-                warnStreamingAttribute(sc, sd, logger);
+                warnStreamingAttribute(cluster, schema.getName(), sd, logger);
             }
         }
     }
 
-    private static void warnStreamingAttribute(StreamingSearchCluster sc, ImmutableSDField sd, DeployLogger logger) {
+    private static void warnStreamingAttribute(String cluster, String schema, ImmutableSDField sd, DeployLogger logger) {
         // If the field is numeric, we can't print this, because we may have converted the field to
         // attribute indexing ourselves (IntegerIndex2Attribute)
         if (sd.getDataType() instanceof NumericDataType) return;
@@ -68,25 +68,25 @@ public class StreamingValidator implements Validator {
             for (var fieldAttribute : sd.getAttributes().values()) {
                 if (fieldAttribute.hnswIndexParams().isPresent()) {
                     logger.logApplicationPackage(Level.WARNING,
-                            "For streaming search cluster '" + sc.getClusterName() +
+                            "For search cluster '" + cluster + "', schema '" + schema +
                                     "', SD field '" + sd.getName() +
                                     "': hnsw index is not relevant and not supported, ignoring setting");
                 }
             }
             return;
         }
-        logger.logApplicationPackage(Level.WARNING, "For streaming search cluster '" + sc.getClusterName() +
+        logger.logApplicationPackage(Level.WARNING, "For search cluster '" + cluster +
                                                     "', SD field '" + sd.getName() +
                                                     "': 'attribute' has same match semantics as 'index'.");
     }
 
-    private static void failStreamingDocumentReferences(Context context, StreamingSearchCluster sc) {
-        for (Attribute attribute : sc.derived().getAttributeFields().attributes()) {
+    private static void failStreamingDocumentReferences(String cluster, DerivedConfiguration derived, Context context) {
+        for (Attribute attribute : derived.getAttributeFields().attributes()) {
             DataType dataType = attribute.getDataType();
             if (dataType instanceof NewDocumentReferenceDataType) {
-                String errorMessage = String.format("For streaming search cluster '%s': Attribute '%s' has type '%s'. " +
+                String errorMessage = String.format("For search cluster '%s', schema '%s': Attribute '%s' has type '%s'. " +
                                                     "Document references and imported fields are not allowed in streaming search.",
-                                                    sc.getClusterName(), attribute.getName(), dataType.getName());
+                                                    cluster, derived.getSchema().getName(), attribute.getName(), dataType.getName());
                 context.illegal(errorMessage);
             }
         }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/application/validation/change/StreamingSearchClusterChangeValidator.java b/config-model/src/main/java/com/yahoo/vespa/model/application/validation/change/StreamingSearchClusterChangeValidator.java
index 3b89467299d..0d42219dade 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/application/validation/change/StreamingSearchClusterChangeValidator.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/application/validation/change/StreamingSearchClusterChangeValidator.java
@@ -1,7 +1,6 @@
 // Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.model.application.validation.change;
 
-import com.yahoo.config.model.api.ConfigChangeAction;
 import com.yahoo.config.model.api.ServiceInfo;
 import com.yahoo.config.provision.ClusterSpec;
 import com.yahoo.documentmodel.NewDocumentType;
@@ -12,11 +11,11 @@ import com.yahoo.vespa.model.application.validation.Validation.ChangeContext;
 import com.yahoo.vespa.model.application.validation.change.search.ChangeMessageBuilder;
 import com.yahoo.vespa.model.application.validation.change.search.DocumentTypeChangeValidator;
 import com.yahoo.vespa.model.content.cluster.ContentCluster;
-import com.yahoo.vespa.model.search.StreamingSearchCluster;
+import com.yahoo.vespa.model.search.DocumentDatabase;
+import com.yahoo.vespa.model.search.SearchCluster;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
@@ -31,39 +30,47 @@ public class StreamingSearchClusterChangeValidator implements ChangeValidator {
         context.previousModel().getContentClusters().forEach((clusterName, currentCluster) -> {
             ContentCluster nextCluster = context.model().getContentClusters().get(clusterName);
             if (nextCluster != null) {
-                List<StreamingSearchCluster> nextStreamingClusters = nextCluster.getSearch().getStreamingClusters();
-                currentCluster.getSearch().getStreamingClusters().forEach(currentStreamingCluster -> {
-                    Optional<StreamingSearchCluster> nextStreamingCluster = findStreamingCluster(currentStreamingCluster.getClusterName(), nextStreamingClusters);
-                    nextStreamingCluster.ifPresent(streamingSearchCluster -> validateStreamingCluster(currentCluster, currentStreamingCluster,
-                            nextCluster, streamingSearchCluster).forEach(context::require));
+                var nextStreamingClusters = nextCluster.getSearch().getClusters();
+                currentCluster.getSearch().getClusters().values().forEach(currentStreamingCluster -> {
+                    SearchCluster nextStreamingCluster = nextStreamingClusters.get(currentStreamingCluster.getClusterName());
+                    if (nextStreamingCluster != null) {
+                        validateStreamingCluster(currentCluster, currentStreamingCluster, nextCluster, nextStreamingCluster).forEach(context::require);
+                    }
                 });
             }
         });
     }
 
-    private static Optional<StreamingSearchCluster> findStreamingCluster(String clusterName, List<StreamingSearchCluster> clusters) {
-        return clusters.stream()
-                .filter(cluster -> cluster.getClusterName().equals(clusterName))
-                .findFirst();
+    private static List<VespaConfigChangeAction> validateStreamingCluster(ContentCluster currentCluster,
+                                                                     SearchCluster currentStreamingCluster,
+                                                                     ContentCluster nextCluster,
+                                                                     SearchCluster nextStreamingCluster) {
+        List<VespaConfigChangeAction> result = new ArrayList<>();
+
+        for (DocumentDatabase currentDB : currentStreamingCluster.getDocumentDbs()) {
+            DocumentDatabase nextDB = nextStreamingCluster.getDocumentDB(currentDB.getName());
+            if (nextDB != null) {
+                result.addAll(validateDocumentDB(currentCluster, currentDB, nextCluster, nextDB));
+            }
+        }
+        return result;
     }
 
-    private static List<ConfigChangeAction> validateStreamingCluster(ContentCluster currentCluster,
-                                                                     StreamingSearchCluster currentStreamingCluster,
-                                                                     ContentCluster nextCluster,
-                                                                     StreamingSearchCluster nextStreamingCluster) {
+    private static List<VespaConfigChangeAction> validateDocumentDB(ContentCluster currentCluster, DocumentDatabase currentDB,
+                                                               ContentCluster nextCluster, DocumentDatabase nextDB) {
         List<VespaConfigChangeAction> result = new ArrayList<>();
 
         result.addAll(validateDocumentTypeChanges(currentCluster.id(),
-                                                  getDocumentType(currentCluster, currentStreamingCluster),
-                                                  getDocumentType(nextCluster, nextStreamingCluster)));
+                getDocumentType(currentCluster, currentDB),
+                getDocumentType(nextCluster, nextDB)));
         result.addAll(validateAttributeFastAccessAdded(currentCluster.id(),
-                                                       currentStreamingCluster.derived().getAttributeFields(),
-                                                       nextStreamingCluster.derived().getAttributeFields()));
+                currentDB.getDerivedConfiguration().getAttributeFields(),
+                nextDB.getDerivedConfiguration().getAttributeFields()));
         result.addAll(validateAttributeFastAccessRemoved(currentCluster.id(),
-                                                         currentStreamingCluster.derived().getAttributeFields(),
-                                                         nextStreamingCluster.derived().getAttributeFields()));
+                currentDB.getDerivedConfiguration().getAttributeFields(),
+                nextDB.getDerivedConfiguration().getAttributeFields()));
 
-        return modifyActions(result, getSearchNodeServices(nextCluster), nextStreamingCluster.getDocTypeName());
+        return modifyActions(result, getSearchNodeServices(nextCluster), nextDB.getName());
     }
 
     private static List<VespaConfigChangeAction> validateDocumentTypeChanges(ClusterSpec.Id id,
@@ -72,8 +79,8 @@ public class StreamingSearchClusterChangeValidator implements ChangeValidator {
         return new DocumentTypeChangeValidator(id, currentDocType, nextDocType).validate();
     }
 
-    private static NewDocumentType getDocumentType(ContentCluster cluster, StreamingSearchCluster streamingCluster) {
-        return cluster.getDocumentDefinitions().get(streamingCluster.getDocTypeName());
+    private static NewDocumentType getDocumentType(ContentCluster cluster, DocumentDatabase db) {
+        return cluster.getDocumentDefinitions().get(db.getName());
     }
 
     private static List<VespaConfigChangeAction> validateAttributeFastAccessAdded(ClusterSpec.Id id,
@@ -110,7 +117,7 @@ public class StreamingSearchClusterChangeValidator implements ChangeValidator {
                 .toList();
     }
 
-    private static List<ConfigChangeAction> modifyActions(List<VespaConfigChangeAction> result,
+    private static List<VespaConfigChangeAction> modifyActions(List<VespaConfigChangeAction> result,
                                                           List<ServiceInfo> services,
                                                           String docTypeName) {
         return result.stream()
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/search/searchchain/LocalProvider.java b/config-model/src/main/java/com/yahoo/vespa/model/container/search/searchchain/LocalProvider.java
index bf331df6e48..75b47fbdf60 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/search/searchchain/LocalProvider.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/search/searchchain/LocalProvider.java
@@ -73,7 +73,7 @@ public class LocalProvider extends Provider implements
 
     //TODO: ugly, restructure this
     private Set<ComponentSpecification> disableStemmingIfStreaming(Set<ComponentSpecification> searcherReferences) {
-        if (!searchCluster.isStreaming()) {
+        if (searchCluster.hasIndexed()) {
             return searcherReferences;
         } else {
             Set<ComponentSpecification> filteredSearcherReferences = new LinkedHashSet<>(searcherReferences);
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index 91a5716b0f8..4d6f454f7e0 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -285,7 +285,7 @@ public class ContentSearchCluster extends TreeConfigProducer<AnyConfigProducer>
         boolean hasStreaming = false;
         boolean hasIndexed = false;
         for (var cluster : clusters.values()) {
-            if (cluster.isStreaming())
+            if (cluster.hasStreaming())
                 hasStreaming = true;
             else
                 hasIndexed = true;
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/search/SearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/search/SearchCluster.java
index db02b7724d7..fc4b96ec384 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/search/SearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/search/SearchCluster.java
@@ -13,9 +13,9 @@ import com.yahoo.vespa.configdefinition.IlscriptsConfig;
 import com.yahoo.config.model.producer.AnyConfigProducer;
 import com.yahoo.config.model.producer.TreeConfigProducer;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
@@ -35,7 +35,7 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
     private Double queryTimeout;
     private Double visibilityDelay = 0.0;
     private final Map<String, SchemaInfo> schemas = new LinkedHashMap<>();
-    private final List<DocumentDatabase> documentDbs = new LinkedList<>();
+    private final Map<String, DocumentDatabase> documentDbs = new LinkedHashMap<>();
 
     public SearchCluster(TreeConfigProducer<?> parent, String clusterName, int index) {
         super(parent, "cluster." + clusterName);
@@ -49,25 +49,19 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
         schemas.put(schema.name(), schema);
     }
     public void add(DocumentDatabase db) {
-        documentDbs.add(db);
+        documentDbs.put(db.getName(), db);
     }
 
     public boolean hasDocumentDB(String name) {
-        for (DocumentDatabase db : documentDbs) {
-            if (db.getName().equals(name)) {
-                return true;
-            }
-        }
-        return false;
+        return documentDbs.containsKey(name);
+    }
+    public DocumentDatabase getDocumentDB(String name) {
+        return documentDbs.get(name);
     }
 
     public String getConfigId(String name) {
-        for (DocumentDatabase db : documentDbs) {
-            if (db.getName().equals(name)) {
-                return db.getConfigId();
-            }
-        }
-        return "";
+        DocumentDatabase db = documentDbs.get(name);
+        return db != null ? db.getConfigId() : "";
     }
 
     /** Returns the schemas that should be active in this cluster. Note: These are added during processing. */
@@ -82,12 +76,17 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
 
     /** Returns the document databases contained in this cluster */
     public List<DocumentDatabase> getDocumentDbs() {
-        return Collections.unmodifiableList(documentDbs);
+        return documentDbs.values().stream().toList();
     }
 
     public String getClusterName()              { return clusterName; }
     public final String getIndexingModeName()   { return getIndexingMode().getName(); }
-    public final boolean isStreaming()          { return getIndexingMode() == IndexingMode.STREAMING; }
+    public final boolean hasStreaming() {
+        return schemas().values().stream().anyMatch(schema -> schema.getIndexMode() == SchemaInfo.IndexMode.STREAMING);
+    }
+    public final boolean hasIndexed() {
+        return schemas().values().stream().anyMatch(schema -> schema.getIndexMode() == SchemaInfo.IndexMode.INDEX);
+    }
 
     public final void setQueryTimeout(Double to) { this.queryTimeout = to; }
     public final void setVisibilityDelay(double delay) { this.visibilityDelay = delay; }
@@ -99,11 +98,9 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
     public final int getClusterIndex() { return index; }
 
     public void fillDocumentDBConfig(String documentType, ProtonConfig.Documentdb.Builder builder) {
-        for (DocumentDatabase sdoc : documentDbs) {
-            if (sdoc.getName().equals(documentType)) {
-                fillDocumentDBConfig(sdoc, builder);
-                return;
-            }
+        DocumentDatabase db = documentDbs.get(documentType);
+        if (db != null) {
+            fillDocumentDBConfig(db, builder);
         }
     }
 
@@ -114,7 +111,7 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
 
     @Override
     public void getConfig(DocumentdbInfoConfig.Builder builder) {
-        for (DocumentDatabase db : documentDbs) {
+        for (DocumentDatabase db : documentDbs.values()) {
             var docDb = new DocumentdbInfoConfig.Documentdb.Builder()
                     .name(db.getName())
                     .mode(db.getDerivedConfiguration().isStreaming()
@@ -125,21 +122,21 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
     }
     @Override
     public void getConfig(IndexInfoConfig.Builder builder) {
-        new Join(documentDbs).getConfig(builder);
+        new Join(documentDbs.values()).getConfig(builder);
     }
 
     @Override
     public void getConfig(SchemaInfoConfig.Builder builder) {
-        new Join(documentDbs).getConfig(builder);
+        new Join(documentDbs.values()).getConfig(builder);
     }
 
     @Override
     public void getConfig(IlscriptsConfig.Builder builder) {
-        new Join(documentDbs).getConfig(builder);
+        new Join(documentDbs.values()).getConfig(builder);
     }
 
     public void getConfig(AttributesConfig.Builder builder) {
-        new Join(documentDbs).getConfig(builder);
+        new Join(documentDbs.values()).getConfig(builder);
     }
 
     @Override
@@ -169,7 +166,7 @@ public abstract class SearchCluster extends TreeConfigProducer<AnyConfigProducer
      * that is handled (by delegating to this) by the {@link IndexedSearchCluster}
      * which is the parent to this. This avoids building the config multiple times.
      */
-    private record Join(List<DocumentDatabase> docDbs) {
+    private record Join(Collection<DocumentDatabase> docDbs) {
 
         public void getConfig(IndexInfoConfig.Builder builder) {
             for (DocumentDatabase docDb : docDbs) {
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/application/validation/StreamingValidatorTest.java b/config-model/src/test/java/com/yahoo/vespa/model/application/validation/StreamingValidatorTest.java
index 5397c30f2bc..ee2d05db7b9 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/application/validation/StreamingValidatorTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/application/validation/StreamingValidatorTest.java
@@ -29,7 +29,7 @@ public class StreamingValidatorTest {
             new VespaModelCreatorWithFilePkg("src/test/cfg/application/validation/document_references_validation/")
                     .create();
         });
-        assertTrue(exception.getMessage().contains("For streaming search cluster 'content.ad': Attribute 'campaign_ref' has type 'Reference<campaign>'. " +
+        assertTrue(exception.getMessage().contains("For search cluster 'content.ad', schema 'ad': Attribute 'campaign_ref' has type 'Reference<campaign>'. " +
                 "Document references and imported fields are not allowed in streaming search."));
     }
 
@@ -52,7 +52,7 @@ public class StreamingValidatorTest {
                     "attribute { distance-metric: euclidean } }");
         var warnings = filter(logger.warnings);
         assertEquals(1, warnings.size());
-        assertEquals("For streaming search cluster 'content.test', SD field 'nn': hnsw index is not relevant and not supported, ignoring setting",
+        assertEquals("For search cluster 'content.test', schema 'test', SD field 'nn': hnsw index is not relevant and not supported, ignoring setting",
                      warnings.get(0));
     }