feat: index contract definitions by asset id when resolving the catalog

core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcher.java

@@ -0,0 +1,126 @@
+/*
+ *  Copyright (c) 2026 Metaform Systems, Inc.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Metaform Systems, Inc. - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.controlplane.catalog;
+
+import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
+import org.eclipse.edc.connector.controlplane.asset.spi.index.AssetIdCompatibility;
+import org.eclipse.edc.connector.controlplane.contract.spi.types.offer.ContractDefinition;
+import org.eclipse.edc.spi.query.Criterion;
+import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import static org.eclipse.edc.connector.controlplane.asset.spi.index.AssetIdCompatibility.ASSET_ID_FIELD;
+import static org.eclipse.edc.spi.query.CriterionOperatorRegistry.EQUAL;
+import static org.eclipse.edc.spi.query.CriterionOperatorRegistry.IN;
+
+/**
+ * Finds the contract definitions whose assets selector matches an asset, for one catalog request.
+ * <p>
+ * The selector predicates are built once. Definitions that select assets by their ids ({@code id = <asset id>} or
+ * {@code id in [<asset ids>]}), the common case, are indexed by those ids, so an asset is only tested against the
+ * definitions indexed under its id and the ones that are not indexed, instead of against all of them. The candidates
+ * are still tested with their predicates, so the result is the same as testing every definition.
+ */
+class ContractDefinitionMatcher {
+
+    private final List<ContractDefinition> definitions;
+    private final List<Predicate<Asset>> predicates;
+    private final Map<String, List<Integer>> indexedByAssetId = new HashMap<>();
+    private final List<Integer> notIndexed = new ArrayList<>();
+
+    ContractDefinitionMatcher(List<ContractDefinition> definitions, CriterionOperatorRegistry criterionOperatorRegistry) {
+        this.definitions = definitions;
+        this.predicates = definitions.stream()
+                .map(definition -> definition.getAssetsSelector().stream()
+                        .map(criterionOperatorRegistry::<Asset>toPredicate)
+                        .reduce(x -> true, Predicate::and))
+                .toList();
+
+        for (var i = 0; i < definitions.size(); i++) {
+            var assetIds = selectedAssetIds(definitions.get(i));
+            if (assetIds == null) {
+                notIndexed.add(i);
+            } else {
+                for (var assetId : assetIds) {
+                    indexedByAssetId.computeIfAbsent(assetId, k -> new ArrayList<>()).add(i);
+                }
+            }
+        }
+    }
+
+    /**
+     * The definitions whose assets selector matches the asset, in the order they were given.
+     */
+    List<ContractDefinition> matching(Asset asset) {
+        var indexed = indexedByAssetId.get(asset.getId());
+        if (indexed == null) {
+            return matching(asset, notIndexed);
+        }
+        var candidates = new ArrayList<Integer>(indexed.size() + notIndexed.size());
+        candidates.addAll(indexed);
+        candidates.addAll(notIndexed);
+        candidates.sort(null);
+        return matching(asset, candidates);
+    }
+
+    private List<ContractDefinition> matching(Asset asset, List<Integer> candidates) {
+        var result = new ArrayList<ContractDefinition>();
+        for (var i : candidates) {
+            if (predicates.get(i).test(asset)) {
+                result.add(definitions.get(i));
+            }
+        }
+        return result;
+    }
+
+    /**
+     * The ids of the assets selected by the definition, if its selector is exactly one {@code id = <asset id>} or
+     * {@code id in [<asset ids>]} criterion with string ids, otherwise {@code null}.
+     */
+    private static Set<String> selectedAssetIds(ContractDefinition definition) {
+        var selector = definition.getAssetsSelector();
+        if (selector.size() != 1 || !isIdOperand(selector.get(0))) {
+            return null;
+        }
+        var criterion = selector.get(0);
+        var operator = criterion.getOperator().toLowerCase();
+        if (EQUAL.equals(operator) && criterion.getOperandRight() instanceof String assetId) {
+            return Set.of(assetId);
+        }
+        if (IN.equals(operator) && criterion.getOperandRight() instanceof Iterable<?> values) {
+            // a set, so that an id listed twice does not make the definition a candidate twice
+            var assetIds = new LinkedHashSet<String>();
+            for (var value : values) {
+                if (!(value instanceof String assetId)) {
+                    return null;
+                }
+                assetIds.add(assetId);
+            }
+            return assetIds;
+        }
+        return null;
+    }
+
+    private static boolean isIdOperand(Criterion criterion) {
+        return ASSET_ID_FIELD.equals(criterion.getOperandLeft()) || AssetIdCompatibility.isLegacyIdOperand(criterion.getOperandLeft());
+    }
+}

core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImpl.java

@@ -35,11 +35,11 @@
 import org.eclipse.edc.spi.query.QuerySpec;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.ArrayList;
 import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import static java.lang.Integer.MAX_VALUE;
@@ -81,11 +81,14 @@ public Stream<Dataset> query(ParticipantContext participantContext, ParticipantA
                 .filter(filterByParticipantContextId(participantContext.getId()))
                 .build();
 
+        var matcher = new ContractDefinitionMatcher(contractDefinitions, criterionOperatorRegistry);
+        // the datasets are only built for the requested page: the offers are enough to know which assets are part of it
         return assetIndex.queryAssets(assetsQuery)
-                .map(asset -> toDataset(contractDefinitions, asset, resolved.policies(), protocol))
-                .filter(Dataset::hasOffers)
+                .map(asset -> new AssetOffers(asset, offers(matcher.matching(asset), resolved.policies())))
+                .filter(AssetOffers::hasOffers)
                 .skip(querySpec.getOffset())
-                .limit(querySpec.getLimit());
+                .limit(querySpec.getLimit())
+                .map(assetOffers -> toDataset(assetOffers, protocol));
     }
 
     @Override
@@ -96,10 +99,12 @@ public Dataset getById(ParticipantContext participantContext, ParticipantAgent a
             return null;
         }
 
+        var matcher = new ContractDefinitionMatcher(contractDefinitions, criterionOperatorRegistry);
         return Optional.of(id)
                 .map(assetId -> assetIndex.findById(participantContext.getId(), assetId))
-                .map(asset -> toDataset(contractDefinitions, asset, resolved.policies(), protocol))
-                .filter(Dataset::hasOffers)
+                .map(asset -> new AssetOffers(asset, offers(matcher.matching(asset), resolved.policies())))
+                .filter(AssetOffers::hasOffers)
+                .map(assetOffers -> toDataset(assetOffers, protocol))
                 .orElse(null);
     }
 
@@ -111,8 +116,8 @@ public Dataset getById(ParticipantContext participantContext, ParticipantAgent a
         var endpointUrl = asset.getPropertyAsString(DCAT_ENDPOINT_URL_ATTRIBUTE);
         if (endpointUrl == null) {
             monitor.warning("""
-                        Asset %s has no 'endpointURL' property and the DataAddress baseUrl is used instead, please adapt it as the
-                        DataAddress will be removed from Asset in the forthcoming versions"""
+                    Asset %s has no 'endpointURL' property and the DataAddress baseUrl is used instead, please adapt it as the
+                    DataAddress will be removed from Asset in the forthcoming versions"""
                     .formatted(asset.getId()));
             endpointUrl = asset.getDataAddress().getStringProperty(EDC_NAMESPACE + "baseUrl", null);
         }
@@ -125,35 +130,46 @@ public Dataset getById(ParticipantContext participantContext, ParticipantAgent a
                         .build());
     }
 
-    private Dataset toDataset(List<ContractDefinition> contractDefinitions, Asset asset, Map<String, Policy> policies, String protocol) {
+    private List<Offer> offers(List<ContractDefinition> contractDefinitions, Map<String, Policy> policies) {
+        var offers = new ArrayList<Offer>();
+        for (var contractDefinition : contractDefinitions) {
+            var policy = policies.computeIfAbsent(contractDefinition.getContractPolicyId(), policyId ->
+                    Optional.ofNullable(policyDefinitionStore.findById(contractDefinition.getParticipantContextId(), policyId))
+                            .map(PolicyDefinition::getPolicy)
+                            .orElse(null)
+            );
+
+            if (policy != null) {
+                offers.add(new Offer(contractDefinition, policy));
+            }
+        }
+        return offers;
+    }
 
+    private Dataset toDataset(AssetOffers assetOffers, String protocol) {
+        var asset = assetOffers.asset();
         var distributions = distributionResolver.getDistributions(protocol, asset);
         var datasetBuilder = buildDataset(asset)
                 .id(asset.getId())
                 .distributions(distributions)
                 .properties(asset.getProperties());
 
-        contractDefinitions.stream()
-                .filter(definition -> definition.getAssetsSelector().stream()
-                        .map(criterionOperatorRegistry::toPredicate)
-                        .reduce(x -> true, Predicate::and)
-                        .test(asset)
-                )
-                .forEach(contractDefinition -> {
-                    var policy = policies.computeIfAbsent(contractDefinition.getContractPolicyId(), policyId ->
-                            Optional.ofNullable(policyDefinitionStore.findById(contractDefinition.getParticipantContextId(), policyId))
-                                    .map(PolicyDefinition::getPolicy)
-                                    .orElse(null)
-                    );
-
-                    if (policy != null) {
-                        var contractId = ContractOfferId.create(contractDefinition.getId(), asset.getId());
-                        var offerPolicy = policy.toBuilder().type(PolicyType.OFFER).build();
-                        datasetBuilder.offer(contractId.toString(), offerPolicy);
-                    }
-                });
+        for (var offer : assetOffers.offers()) {
+            var contractId = ContractOfferId.create(offer.contractDefinition().getId(), asset.getId());
+            var offerPolicy = offer.policy().toBuilder().type(PolicyType.OFFER).build();
+            datasetBuilder.offer(contractId.toString(), offerPolicy);
+        }
 
         return datasetBuilder.build();
     }
 
+    private record Offer(ContractDefinition contractDefinition, Policy policy) {
+    }
+
+    private record AssetOffers(Asset asset, List<Offer> offers) {
+        boolean hasOffers() {
+            return !offers.isEmpty();
+        }
+    }
+
 }

core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcherTest.java

@@ -0,0 +1,117 @@
+/*
+ *  Copyright (c) 2026 Metaform Systems, Inc.
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Metaform Systems, Inc. - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.controlplane.catalog;
+
+import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
+import org.eclipse.edc.connector.controlplane.contract.spi.types.offer.ContractDefinition;
+import org.eclipse.edc.connector.controlplane.query.asset.AssetPropertyLookup;
+import org.eclipse.edc.query.CriterionOperatorRegistryImpl;
+import org.eclipse.edc.spi.query.Criterion;
+import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.spi.constants.CoreConstants.EDC_NAMESPACE;
+import static org.mockito.Mockito.mock;
+
+class ContractDefinitionMatcherTest {
+
+    private final CriterionOperatorRegistry criterionOperatorRegistry = registry();
+
+    @Test
+    void matching_shouldReturnDefinitionsSelectingTheAssetById() {
+        var definitions = List.of(
+                definition("cd-1", new Criterion("id", "=", "asset-1")),
+                definition("cd-2", new Criterion(EDC_NAMESPACE + "id", "=", "asset-2")),
+                definition("cd-3", new Criterion("id", "=", "asset-1")));
+
+        var matcher = new ContractDefinitionMatcher(definitions, criterionOperatorRegistry);
+
+        assertThat(matcher.matching(asset("asset-1"))).extracting(ContractDefinition::getId).containsExactly("cd-1", "cd-3");
+        assertThat(matcher.matching(asset("asset-2"))).extracting(ContractDefinition::getId).containsExactly("cd-2");
+        assertThat(matcher.matching(asset("asset-3"))).isEmpty();
+    }
+
+    @Test
+    void matching_shouldReturnDefinitionsSelectingTheAssetByIdIn() {
+        var definitions = List.of(
+                definition("cd-1", new Criterion("id", "in", List.of("asset-1", "asset-2"))),
+                definition("cd-2", new Criterion(EDC_NAMESPACE + "id", "IN", List.of("asset-2", "asset-3"))));
+
+        var matcher = new ContractDefinitionMatcher(definitions, criterionOperatorRegistry);
+
+        assertThat(matcher.matching(asset("asset-1"))).extracting(ContractDefinition::getId).containsExactly("cd-1");
+        assertThat(matcher.matching(asset("asset-2"))).extracting(ContractDefinition::getId).containsExactly("cd-1", "cd-2");
+        assertThat(matcher.matching(asset("asset-4"))).isEmpty();
+    }
+
+    @Test
+    void matching_shouldReturnDefinitionOnce_whenIdListedTwice() {
+        var definitions = List.of(definition("cd", new Criterion("id", "in", List.of("asset-1", "asset-1"))));
+
+        var matcher = new ContractDefinitionMatcher(definitions, criterionOperatorRegistry);
+
+        assertThat(matcher.matching(asset("asset-1"))).extracting(ContractDefinition::getId).containsExactly("cd");
+    }
+
+    @Test
+    void matching_shouldTestDefinitionsNotSelectingById_againstEveryAsset() {
+        var definitions = List.of(
+                definition("all"),
+                definition("by-property", new Criterion("type", "=", "dataset")),
+                definition("by-id-and-property", new Criterion("id", "=", "asset-1"), new Criterion("type", "=", "other")),
+                definition("by-id-not-in", new Criterion("id", "not in", List.of("asset-1"))),
+                definition("by-id-in-with-number", new Criterion("id", "in", List.of("asset-1", 3))));
+
+        var matcher = new ContractDefinitionMatcher(definitions, criterionOperatorRegistry);
+
+        assertThat(matcher.matching(asset("asset-1"))).extracting(ContractDefinition::getId).containsExactly("all", "by-property", "by-id-in-with-number");
+        assertThat(matcher.matching(asset("asset-3"))).extracting(ContractDefinition::getId).containsExactly("all", "by-property", "by-id-not-in");
+    }
+
+    @Test
+    void matching_shouldKeepTheOrderOfTheDefinitions() {
+        var definitions = List.of(
+                definition("all"),
+                definition("by-id", new Criterion("id", "=", "asset-1")),
+                definition("by-property", new Criterion("type", "=", "dataset")));
+
+        var matcher = new ContractDefinitionMatcher(definitions, criterionOperatorRegistry);
+
+        assertThat(matcher.matching(asset("asset-1"))).extracting(ContractDefinition::getId).containsExactly("all", "by-id", "by-property");
+    }
+
+    private static CriterionOperatorRegistry registry() {
+        var registry = CriterionOperatorRegistryImpl.ofDefaults();
+        registry.registerPropertyLookup(new AssetPropertyLookup(mock()));
+        return registry;
+    }
+
+    private static ContractDefinition definition(String id, Criterion... selector) {
+        return ContractDefinition.Builder.newInstance()
+                .id(id)
+                .participantContextId("participantContextId")
+                .accessPolicyId("access")
+                .contractPolicyId("contract")
+                .assetsSelector(List.of(selector))
+                .build();
+    }
+
+    private static Asset asset(String id) {
+        return Asset.Builder.newInstance().id(id).property("type", "dataset").build();
+    }
+}