diff --git a/core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcher.java b/core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcher.java new file mode 100644 index 00000000000..ce4ecb6cd08 --- /dev/null +++ b/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. + *

+ * The selector predicates are built once. Definitions that select assets by their ids ({@code id = } or + * {@code id in []}), 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 definitions; + private final List> predicates; + private final Map> indexedByAssetId = new HashMap<>(); + private final List notIndexed = new ArrayList<>(); + + ContractDefinitionMatcher(List definitions, CriterionOperatorRegistry criterionOperatorRegistry) { + this.definitions = definitions; + this.predicates = definitions.stream() + .map(definition -> definition.getAssetsSelector().stream() + .map(criterionOperatorRegistry::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 matching(Asset asset) { + var indexed = indexedByAssetId.get(asset.getId()); + if (indexed == null) { + return matching(asset, notIndexed); + } + var candidates = new ArrayList(indexed.size() + notIndexed.size()); + candidates.addAll(indexed); + candidates.addAll(notIndexed); + candidates.sort(null); + return matching(asset, candidates); + } + + private List matching(Asset asset, List candidates) { + var result = new ArrayList(); + 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 = } or + * {@code id in []} criterion with string ids, otherwise {@code null}. + */ + private static Set 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(); + 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()); + } +} diff --git a/core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImpl.java b/core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImpl.java index 6b0d40eee77..53daa9f9c2f 100644 --- a/core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImpl.java +++ b/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 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 contractDefinitions, Asset asset, Map policies, String protocol) { + private List offers(List contractDefinitions, Map policies) { + var offers = new ArrayList(); + 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 offers) { + boolean hasOffers() { + return !offers.isEmpty(); + } + } + } diff --git a/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcherTest.java b/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/ContractDefinitionMatcherTest.java new file mode 100644 index 00000000000..f8ce4ac702b --- /dev/null +++ b/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(); + } +} diff --git a/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImplTest.java b/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImplTest.java index 297007835f4..1f62d13ea86 100644 --- a/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImplTest.java +++ b/core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImplTest.java @@ -64,6 +64,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; class DatasetResolverImplTest { @@ -224,6 +225,23 @@ void shouldLimitDataset_whenSingleDefinitionAndMultipleAssets_contained() { assertThat(datasets).hasSize(3).map(getId()).containsExactly("2", "3", "4"); } + @Test + void shouldBuildDatasetsOnlyForTheRequestedPage() { + var contractDefinition = contractDefinitionBuilder("definitionId").contractPolicyId("contractPolicyId").build(); + var assets = range(0, 10).mapToObj(it -> createAsset(String.valueOf(it)).build()).toList(); + when(definitionResolver.resolveFor(any(), any())).thenReturn(new ResolvedContractDefinitions(List.of(contractDefinition))); + when(assetIndex.queryAssets(isA(QuerySpec.class))).thenAnswer(i -> assets.stream()); + when(policyStore.findById("participantContextId", "contractPolicyId")).thenReturn(PolicyDefinition.Builder.newInstance().policy(Policy.Builder.newInstance().build()).build()); + var querySpec = QuerySpec.Builder.newInstance().range(new Range(6, 8)).build(); + + var datasets = datasetResolver.query(createParticipantContext(), createParticipantAgent(), querySpec, "protocol").toList(); + + assertThat(datasets).map(getId()).containsExactly("6", "7"); + verify(distributionResolver).getDistributions("protocol", assets.get(6)); + verify(distributionResolver).getDistributions("protocol", assets.get(7)); + verifyNoMoreInteractions(distributionResolver); + } + @Test void shouldLimitDataset_whenSingleDefinitionAndMultipleAssets_overflowing() { var contractDefinition = contractDefinitionBuilder("definitionId").contractPolicyId("contractPolicyId").build();