feat: filter state by string

core/common/lib/sql-lib/src/main/java/org/eclipse/edc/sql/translation/EntityStateFieldTranslator.java

@@ -0,0 +1,43 @@
+/*
+ *  Copyright (c) 2025 Cofinity-X
+ *
+ *  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:
+ *       Cofinity-X - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.sql.translation;
+
+import org.eclipse.edc.spi.entity.StateResolver;
+import org.eclipse.edc.spi.query.Criterion;
+
+import java.util.Collection;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Supports the string representation of a state, that will be converted into int by the {@link #stateResolver}.
+ */
+public class EntityStateFieldTranslator extends PlainColumnFieldTranslator {
+
+    private final StateResolver stateResolver;
+
+    public EntityStateFieldTranslator(String columnName, StateResolver stateResolver) {
+        super(columnName);
+        this.stateResolver = stateResolver;
+    }
+
+    @Override
+    public Collection<Object> toParameters(Criterion criterion) {
+        return super.toParameters(criterion)
+                .stream()
+                .map(it -> it instanceof String stringState ? stateResolver.resolve(stringState) : it)
+                .collect(toList());
+    }
+}

core/common/lib/sql-lib/src/main/java/org/eclipse/edc/sql/translation/FieldTranslator.java

@@ -50,14 +50,7 @@ public interface FieldTranslator {
      */
     WhereClause toWhereClause(List<PathItem> path, Criterion criterion, SqlOperator operator);
 
-    static String toValuePlaceholder(Criterion criterion) {
-        if (criterion.getOperandRight() instanceof Collection<?> collection) {
-            return format("(%s)", String.join(",", nCopies(collection.size(), PREPARED_STATEMENT_PLACEHOLDER)));
-        }
-        return PREPARED_STATEMENT_PLACEHOLDER;
-    }
-
-    static Collection<Object> toParameters(Criterion criterion) {
+    default Collection<Object> toParameters(Criterion criterion) {
         var operandRight = criterion.getOperandRight();
         if (operandRight == null) {
             return emptyList();
@@ -68,4 +61,11 @@ static Collection<Object> toParameters(Criterion criterion) {
         }
     }
 
+    static String toValuePlaceholder(Criterion criterion) {
+        if (criterion.getOperandRight() instanceof Collection<?> collection) {
+            return format("(%s)", String.join(",", nCopies(collection.size(), PREPARED_STATEMENT_PLACEHOLDER)));
+        }
+        return PREPARED_STATEMENT_PLACEHOLDER;
+    }
+
 }

core/common/lib/sql-lib/src/main/java/org/eclipse/edc/sql/translation/JsonArrayTranslator.java

@@ -20,7 +20,6 @@
 import java.util.Collection;
 import java.util.List;
 
-import static org.eclipse.edc.sql.translation.FieldTranslator.toParameters;
 import static org.eclipse.edc.sql.translation.FieldTranslator.toValuePlaceholder;
 
 /**

core/common/lib/sql-lib/src/main/java/org/eclipse/edc/sql/translation/JsonFieldTranslator.java

@@ -22,7 +22,6 @@
 
 import static java.lang.String.format;
 import static java.util.stream.IntStream.range;
-import static org.eclipse.edc.sql.translation.FieldTranslator.toParameters;
 import static org.eclipse.edc.sql.translation.FieldTranslator.toValuePlaceholder;
 
 public class JsonFieldTranslator implements FieldTranslator {

core/common/lib/sql-lib/src/main/java/org/eclipse/edc/sql/translation/PlainColumnFieldTranslator.java

@@ -19,7 +19,6 @@
 
 import java.util.List;
 
-import static org.eclipse.edc.sql.translation.FieldTranslator.toParameters;
 import static org.eclipse.edc.sql.translation.FieldTranslator.toValuePlaceholder;
 
 public class PlainColumnFieldTranslator implements FieldTranslator {

core/common/lib/sql-lib/src/test/java/org/eclipse/edc/sql/translation/EntityStateFieldTranslatorTest.java

@@ -0,0 +1,85 @@
+/*
+ *  Copyright (c) 2025 Cofinity-X
+ *
+ *  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:
+ *       Cofinity-X - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.sql.translation;
+
+import org.eclipse.edc.spi.entity.StateResolver;
+import org.eclipse.edc.util.reflection.PathItem;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.spi.query.Criterion.criterion;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class EntityStateFieldTranslatorTest {
+
+    private final StateResolver stateResolver = mock();
+    private final EntityStateFieldTranslator translator = new EntityStateFieldTranslator("column_name", stateResolver);
+
+    @Test
+    void shouldReturnWhereClause_whenOperatorSpecified() {
+        var path = PathItem.parse("any");
+
+        var result = translator.toWhereClause(path, criterion("field", "=", 100), createOperator());
+
+        assertThat(result.sql()).isEqualTo("column_name any ?");
+        assertThat(result.parameters()).containsExactly(100);
+    }
+
+    @Test
+    void shouldMapMultipleParameters_whenRightOperandIsCollection() {
+        var path = PathItem.parse("any");
+
+        var result = translator.toWhereClause(path, criterion("field", "in", List.of(100, 200)), createOperator());
+
+        assertThat(result.sql()).isEqualTo("column_name any (?,?)");
+        assertThat(result.parameters()).containsExactly(100, 200);
+    }
+
+    @Test
+    void shouldTranslateStateToCode_whenInputTypeIsString() {
+        when(stateResolver.resolve(any())).thenReturn(100);
+        var path = PathItem.parse("any");
+
+        var result = translator.toWhereClause(path, criterion("field", "=", "STATE"), createOperator());
+
+        assertThat(result.sql()).isEqualTo("column_name any ?");
+        assertThat(result.parameters()).containsExactly(100);
+        verify(stateResolver).resolve("STATE");
+    }
+
+    @Test
+    void shouldMapMultipleParametersToCode_whenRightOperandIsStringCollection() {
+        when(stateResolver.resolve(any())).thenReturn(100).thenReturn(200);
+        var path = PathItem.parse("any");
+
+        var result = translator.toWhereClause(path, criterion("field", "in", List.of("STATE", "ANOTHER_STATE")), createOperator());
+
+        assertThat(result.sql()).isEqualTo("column_name any (?,?)");
+        assertThat(result.parameters()).containsExactly(100, 200);
+        verify(stateResolver).resolve("STATE");
+        verify(stateResolver).resolve("ANOTHER_STATE");
+    }
+
+    @NotNull
+    private static SqlOperator createOperator() {
+        return new SqlOperator("any", Object.class);
+    }
+}

core/common/lib/store-lib/src/main/java/org/eclipse/edc/store/AndOperatorCriteriaToPredicate.java

@@ -0,0 +1,44 @@
+/*
+ *  Copyright (c) 2025 Cofinity-X
+ *
+ *  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:
+ *       Cofinity-X - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.store;
+
+import org.eclipse.edc.spi.query.CriteriaToPredicate;
+import org.eclipse.edc.spi.query.Criterion;
+import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+/**
+ * Converts criteria to a Predicate that converts all the criterion to predicate using the passed {@link #criterionOperatorRegistry}
+ * and joins them with "and" operator.
+ *
+ * @param <T> the object type
+ */
+public class AndOperatorCriteriaToPredicate<T> implements CriteriaToPredicate<T> {
+
+    private final CriterionOperatorRegistry criterionOperatorRegistry;
+
+    public AndOperatorCriteriaToPredicate(CriterionOperatorRegistry criterionOperatorRegistry) {
+        this.criterionOperatorRegistry = criterionOperatorRegistry;
+    }
+
+    @Override
+    public Predicate<T> convert(List<Criterion> criteria) {
+        return criteria.stream()
+                .map(criterionOperatorRegistry::<T>toPredicate)
+                .reduce(x -> true, Predicate::and);
+    }
+}

core/common/lib/store-lib/src/main/java/org/eclipse/edc/store/InMemoryStatefulEntityStore.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.edc.store;
 
+import org.eclipse.edc.spi.entity.StateResolver;
 import org.eclipse.edc.spi.entity.StatefulEntity;
 import org.eclipse.edc.spi.persistence.Lease;
 import org.eclipse.edc.spi.persistence.StateEntityStore;
@@ -55,8 +56,8 @@ public class InMemoryStatefulEntityStore<T extends StatefulEntity<T>> implements
     private final Map<String, Lease> leases = new HashMap<>();
     protected final CriterionOperatorRegistry criterionOperatorRegistry;
 
-    public InMemoryStatefulEntityStore(Class<T> clazz, String lockId, Clock clock, CriterionOperatorRegistry criterionOperatorRegistry) {
-        queryResolver = new ReflectionBasedQueryResolver<>(clazz, criterionOperatorRegistry);
+    public InMemoryStatefulEntityStore(Class<T> clazz, String lockId, Clock clock, CriterionOperatorRegistry criterionOperatorRegistry, StateResolver stateResolver) {
+        this.queryResolver = new ReflectionBasedQueryResolver<>(clazz, new StatefulEntityCriteriaToPredicate<>(criterionOperatorRegistry, stateResolver));
         this.lockId = lockId;
         this.clock = clock;
         this.criterionOperatorRegistry = criterionOperatorRegistry;

core/common/lib/store-lib/src/main/java/org/eclipse/edc/store/ReflectionBasedQueryResolver.java

@@ -14,6 +14,7 @@
 
 package org.eclipse.edc.store;
 
+import org.eclipse.edc.spi.query.CriteriaToPredicate;
 import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
 import org.eclipse.edc.spi.query.QueryResolver;
 import org.eclipse.edc.spi.query.QuerySpec;
@@ -22,8 +23,6 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.util.Comparator;
-import java.util.function.BinaryOperator;
-import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import static java.lang.String.format;
@@ -36,7 +35,7 @@
 public class ReflectionBasedQueryResolver<T> implements QueryResolver<T> {
 
     private final Class<T> typeParameterClass;
-    private final CriterionOperatorRegistry criterionOperatorRegistry;
+    private final CriteriaToPredicate<T> criteriaToPredicate;
 
     /**
      * Constructor for ReflectionBasedQueryResolver
@@ -45,10 +44,13 @@ public class ReflectionBasedQueryResolver<T> implements QueryResolver<T> {
      * @param criterionOperatorRegistry converts from a criterion to a predicate
      */
     public ReflectionBasedQueryResolver(Class<T> typeParameterClass, CriterionOperatorRegistry criterionOperatorRegistry) {
-        this.typeParameterClass = typeParameterClass;
-        this.criterionOperatorRegistry = criterionOperatorRegistry;
+        this(typeParameterClass, new AndOperatorCriteriaToPredicate<>(criterionOperatorRegistry));
     }
 
+    public ReflectionBasedQueryResolver(Class<T> typeParameterClass, CriteriaToPredicate<T> criteriaToPredicate) {
+        this.typeParameterClass = typeParameterClass;
+        this.criteriaToPredicate = criteriaToPredicate;
+    }
 
     /**
      * Method to query a stream by provided specification.
@@ -58,16 +60,13 @@ public ReflectionBasedQueryResolver(Class<T> typeParameterClass, CriterionOperat
      *
      * @param stream      stream to be queried.
      * @param spec        query specification.
-     * @param accumulator accumulation operation, e.g. Predicate::and, Predicate::or, etc.
      * @return stream result from queries.
      */
     @Override
-    public Stream<T> query(Stream<T> stream, QuerySpec spec, BinaryOperator<Predicate<Object>> accumulator, Predicate<Object> fallback) {
-        var andPredicate = spec.getFilterExpression().stream()
-                .map(criterionOperatorRegistry::toPredicate)
-                .reduce(fallback, accumulator);
+    public Stream<T> query(Stream<T> stream, QuerySpec spec) {
+        var predicate = criteriaToPredicate.convert(spec.getFilterExpression());
 
-        var filteredStream = stream.filter(andPredicate);
+        var filteredStream = stream.filter(predicate);
 
         // sort
         var sortField = spec.getSortField();

core/common/lib/store-lib/src/main/java/org/eclipse/edc/store/StatefulEntityCriteriaToPredicate.java

@@ -0,0 +1,55 @@
+/*
+ *  Copyright (c) 2025 Cofinity-X
+ *
+ *  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:
+ *       Cofinity-X - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.store;
+
+import org.eclipse.edc.spi.entity.StateResolver;
+import org.eclipse.edc.spi.entity.StatefulEntity;
+import org.eclipse.edc.spi.query.CriteriaToPredicate;
+import org.eclipse.edc.spi.query.Criterion;
+import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+import static org.eclipse.edc.spi.query.Criterion.criterion;
+
+/**
+ * Convert criteria to predicate for stateful entities.
+ *
+ * @param <E> entity type
+ */
+public class StatefulEntityCriteriaToPredicate<E extends StatefulEntity<E>> implements CriteriaToPredicate<E> {
+
+    private final CriterionOperatorRegistry criterionOperatorRegistry;
+    private final StateResolver stateResolver;
+
+    public StatefulEntityCriteriaToPredicate(CriterionOperatorRegistry criterionOperatorRegistry, StateResolver stateResolver) {
+        this.criterionOperatorRegistry = criterionOperatorRegistry;
+        this.stateResolver = stateResolver;
+    }
+
+    @Override
+    public Predicate<E> convert(List<Criterion> criteria) {
+        return criteria.stream()
+                .map(criterion -> {
+                    if (criterion.getOperandLeft().equals("state") && criterion.getOperandRight() instanceof String stateString) {
+                        return criterion(criterion.getOperandLeft(), criterion.getOperator(), stateResolver.resolve(stateString));
+                    }
+                    return criterion;
+                })
+                .map(criterionOperatorRegistry::<E>toPredicate)
+                .reduce(x -> true, Predicate::and);
+    }
+}

core/common/lib/store-lib/src/test/java/org/eclipse/edc/store/ReflectionBasedQueryResolverTest.java

@@ -25,7 +25,6 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
-import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -57,16 +56,6 @@ void verifyQuery_equalStringProperty() {
 
     }
 
-    @Test
-    void verifyQuery_equalStringProperty_accumulateOr() {
-        var stream = Stream.concat(
-                IntStream.range(0, 5).mapToObj(i -> new FakeItem(i, "Alice")),
-                IntStream.range(5, 10).mapToObj(i -> new FakeItem(i, "Bob")));
-
-        var spec = QuerySpec.Builder.newInstance().filter(List.of(criterion("name", "=", "Alice"), criterion("name", "=", "Bob"))).build();
-        assertThat(queryResolver.query(stream, spec, Predicate::or, x -> false)).hasSize(10).extracting(FakeItem::getName).containsOnly("Bob", "Alice");
-    }
-
     @Test
     void verifyQuery_criterionFilterIntProperty() {
         var stream = Stream.concat(