[MODFQMMGR-1188] Support MARC indicator + subfield querying

NEWS.md

@@ -1,6 +1,7 @@
 # 4.2.x - Umbrellaleaf
 
 ## 4.2.0 (Next)
+- [MODFQMMGR-1188](https://folio-org.atlassian.net/browse/MODFQMMGR-1188) Support MARC indicator + subfield (constrained-subfield) querying on the SRS record entity type
 - [MODFQMMGR-1143](https://folio-org.atlassian.net/browse/MODFQMMGR-1143) Support MARC indicator querying on the SRS record entity type
 - [MODFQMMGR-1141](https://folio-org.atlassian.net/browse/MODFQMMGR-1141) Support top-level MARC tag and control-field querying on the SRS record entity type
 - [MODFQMMGR-1142](https://folio-org.atlassian.net/browse/MODFQMMGR-1142) Support dynamic MARC subfield querying on the SRS record entity type

src/main/java/org/folio/fqm/service/FqlToSqlConverterService.java

@@ -413,7 +413,7 @@ private static Condition handleContains(ContainsCondition containsCondition, Ent
 
   private static Condition handleMarcCondition(FieldCondition<?> fieldCondition, EntityType entityType,
                                                MarcQueryContext marcQueryContext) {
-    if (marcQueryContext.marcField().isIndicator()) {
+    if (marcQueryContext.marcField().isIndicatorTarget()) {
       return handleMarcIndicatorCondition(fieldCondition, entityType, marcQueryContext);
     }
     return switch (fieldCondition) {

src/main/java/org/folio/fqm/utils/MarcFieldFactory.java

@@ -35,6 +35,14 @@ public class MarcFieldFactory {
   // Indicator form (e.g. marc_245_ind1 / marc_245_ind2). Targets the ind1/ind2 column of the tag rather than
   // a subfield value. Only valid for data-field tags (010+); control fields have no indicators.
   private static final Pattern INDICATOR_PATTERN = Pattern.compile("^marc_(?<tag>\\d{3})_ind(?<indicator>[12])$", Pattern.CASE_INSENSITIVE);
+  // Constrained-subfield form (e.g. marc_245_ind1_7_a, marc_245_ind1_blank_a). Targets a subfield value like
+  // the subfield form, but with the indicator fixed to a constant matched on the SAME marc_indexers row. The
+  // fixed indicator value is a single alphanumeric or the public token "blank". Data-field tags (010+) only.
+  private static final Pattern CONSTRAINED_SUBFIELD_PATTERN = Pattern.compile(
+    "^marc_(?<tag>\\d{3})_ind(?<indicator>[12])_(?<indValue>blank|[a-z0-9])_(?<subfield>[a-z0-9])$",
+    Pattern.CASE_INSENSITIVE);
+  private static final String BLANK_INDICATOR_TOKEN = "blank";
+  private static final String BLANK_INDICATOR_STORAGE = "#";
   // Generic scanner for "fieldName": keys in a raw FQL query. It intentionally does NOT encode the MARC
   // grammar; every candidate key is validated through parse()/isMarcFieldName so the grammar lives in
   // exactly one place and the two cannot drift.
@@ -161,36 +169,50 @@ public static Optional<MarcQueryContext> createQueryContext(EntityType entityTyp
   }
 
   public static Optional<MarcFieldName> parse(String fieldName) {
+    // Control fields (001-009) have no subfields or indicators, so only the tag-only form below is valid for
+    // them. The subfield/constrained/indicator forms are rejected for control tags.
     Matcher subfieldMatcher = SUBFIELD_PATTERN.matcher(fieldName);
-    // Control fields (001-009) carry a single string value with no subfields, so the subfield form is only
-    // valid for data-field tags (010+). Control fields are queryable via the tag-only form below.
     if (subfieldMatcher.matches() && !isControlFieldTag(subfieldMatcher.group("tag"))) {
       // Preserve the original field name so the synthesized column matches the name referenced in the query,
       // but normalize the subfield code to lower case to match how it is stored in marc_indexers.
       return Optional.of(new MarcFieldName(
         fieldName,
         subfieldMatcher.group("tag"),
         subfieldMatcher.group("subfield").toLowerCase(),
+        null,
         null
       ));
     }
 
+    Matcher constrainedMatcher = CONSTRAINED_SUBFIELD_PATTERN.matcher(fieldName);
+    if (constrainedMatcher.matches() && !isControlFieldTag(constrainedMatcher.group("tag"))) {
+      // Subfield value target, with the indicator fixed to a constant (blank -> '#', lower-cased) matched on
+      // the same row.
+      return Optional.of(new MarcFieldName(
+        fieldName,
+        constrainedMatcher.group("tag"),
+        constrainedMatcher.group("subfield").toLowerCase(),
+        constrainedMatcher.group("indicator"),
+        normalizeIndicatorValue(constrainedMatcher.group("indValue"))
+      ));
+    }
+
     Matcher indicatorMatcher = INDICATOR_PATTERN.matcher(fieldName);
-    // Control fields have no indicators, so the indicator form is only valid for data-field tags (010+).
     if (indicatorMatcher.matches() && !isControlFieldTag(indicatorMatcher.group("tag"))) {
       return Optional.of(new MarcFieldName(
         fieldName,
         indicatorMatcher.group("tag"),
         null,
-        indicatorMatcher.group("indicator")
+        indicatorMatcher.group("indicator"),
+        null
       ));
     }
 
     Matcher tagMatcher = TAG_PATTERN.matcher(fieldName);
     if (tagMatcher.matches()) {
       // Tag-only: no subfield target, so the predicate matches any subfield of the tag (and is the only
       // valid form for control fields, which have no subfields or indicators).
-      return Optional.of(new MarcFieldName(fieldName, tagMatcher.group("tag"), null, null));
+      return Optional.of(new MarcFieldName(fieldName, tagMatcher.group("tag"), null, null, null));
     }
 
     return Optional.empty();
@@ -202,6 +224,13 @@ private static boolean isControlFieldTag(String tag) {
     return tag.startsWith("00");
   }
 
+  // Normalizes a fixed indicator value from a constrained-subfield field name: the public token "blank" maps
+  // to the stored '#', and other (single alphanumeric) values are lower-cased so the constraint matches
+  // case-insensitively.
+  private static String normalizeIndicatorValue(String rawValue) {
+    return BLANK_INDICATOR_TOKEN.equalsIgnoreCase(rawValue) ? BLANK_INDICATOR_STORAGE : rawValue.toLowerCase();
+  }
+
   public static Optional<EntityTypeColumn> findMarcPlaceholder(EntityType entityType) {
     return entityType.getColumns().stream()
       .filter(MarcFieldFactory::isGenericMarcPlaceholder)
@@ -225,13 +254,13 @@ private static String buildValueGetter(MarcFieldName marcField, String marcIdGet
     // once per subfield (e.g. a 245 with $a$b yields ["1","1"]). DISTINCT collapses that artifactual
     // duplication to the distinct indicator value(s). Subfield/tag values are aggregated as-is, since their
     // repetition is meaningful.
-    String distinct = marcField.isIndicator() ? "DISTINCT " : "";
+    String distinct = marcField.isIndicatorTarget() ? "DISTINCT " : "";
     return """
       (
         SELECT jsonb_agg(%smarc.%s) FILTER (WHERE marc.%s IS NOT NULL)
         FROM %s marc
         WHERE marc.marc_id = %s
-          AND marc.field_no = '%s'%s
+          AND marc.field_no = '%s'%s%s
       )
     """.formatted(
       distinct,
@@ -240,6 +269,7 @@ SELECT jsonb_agg(%smarc.%s) FILTER (WHERE marc.%s IS NOT NULL)
       interpolateTenant(MARC_INDEXERS_VIEW, tenantId),
       marcIdGetter,
       marcField.tag(),
+      marcField.indicatorConstraintClause(),
       marcField.subfieldClause()
     ).trim();
   }
@@ -258,34 +288,53 @@ private static Optional<String> extractMarcTableName(String valueGetter) {
   }
 
   /**
-   * A parsed MARC field reference. Exactly one optional target is set: {@code subfield} for the subfield
-   * form, {@code indicator} ("1"/"2") for the indicator form; both null is the tag-only form.
+   * A parsed MARC field reference across the supported forms:
+   * <ul>
+   *   <li>tag-only: {@code subfield}, {@code indicatorNumber}, {@code indicatorValue} all null</li>
+   *   <li>subfield: {@code subfield} set</li>
+   *   <li>indicator-only: {@code indicatorNumber} ("1"/"2") set, {@code indicatorValue} null (targets ind)</li>
+   *   <li>constrained-subfield: {@code indicatorNumber} + {@code indicatorValue} (fixed) + {@code subfield}</li>
+   * </ul>
    */
-  public record MarcFieldName(String fieldName, String tag, String subfield, String indicator) {
+  public record MarcFieldName(String fieldName, String tag, String subfield, String indicatorNumber, String indicatorValue) {
 
-    public boolean isIndicator() {
-      return indicator != null;
+    /** True only for the indicator-only form, where the query targets the indicator column itself. When a
+     *  fixed {@code indicatorValue} is present the indicator is a constraint and the subfield is the target. */
+    public boolean isIndicatorTarget() {
+      return indicatorNumber != null && indicatorValue == null;
     }
 
     public String labelAlias() {
       // Prefixed with "MARC" so the label identifies it as a MARC field (consistent with the generic "MARC"
-      // placeholder), e.g. "MARC 245" (tag-only), "MARC 245$a" (subfield), "MARC 245 ind1" (indicator).
-      if (isIndicator()) {
-        return "MARC %s ind%s".formatted(tag, indicator);
+      // placeholder), e.g. "MARC 245" (tag-only), "MARC 245$a" (subfield), "MARC 245 ind1" (indicator),
+      // "MARC 245 ind1=7 $a" (constrained subfield).
+      if (isIndicatorTarget()) {
+        return "MARC %s ind%s".formatted(tag, indicatorNumber);
+      }
+      if (indicatorValue != null) {
+        // Show the public "blank" token in the label rather than the stored '#'.
+        String displayValue = BLANK_INDICATOR_STORAGE.equals(indicatorValue) ? BLANK_INDICATOR_TOKEN : indicatorValue;
+        return "MARC %s ind%s=%s $%s".formatted(tag, indicatorNumber, displayValue, subfield);
       }
       return subfield == null ? "MARC %s".formatted(tag) : "MARC %s$%s".formatted(tag, subfield);
     }
 
-    /** The marc_indexers column this field targets: ind1/ind2 for indicators, otherwise the subfield value. */
+    /** The marc_indexers column this field targets: ind1/ind2 for indicator-only, otherwise the subfield value. */
     public String targetColumn() {
-      return isIndicator() ? "ind" + indicator : "value";
+      return isIndicatorTarget() ? "ind" + indicatorNumber : "value";
     }
 
-    /** WHERE fragment narrowing to a specific subfield; empty for tag-only and indicator fields. */
+    /** WHERE fragment narrowing to a specific subfield; empty for tag-only and indicator-only fields. */
     public String subfieldClause() {
       return subfield == null ? "" : " AND marc.subfield_no = '%s'".formatted(subfield);
     }
 
+    /** WHERE fragment fixing the indicator to a constant (constrained-subfield form); empty otherwise. Matched
+     *  case-insensitively, consistent with indicator matching. */
+    public String indicatorConstraintClause() {
+      return indicatorValue == null ? "" : " AND lower(marc.ind%s) = '%s'".formatted(indicatorNumber, indicatorValue);
+    }
+
     public String filterValueGetter() {
       return "lower(marc.%s)".formatted(targetColumn());
     }
@@ -304,6 +353,9 @@ public String filterValueGetter() {
 
     public String whereClause() {
       String clause = "marc.marc_id = %s and marc.field_no = '%s'".formatted(marcIdGetter, marcField.tag());
+      if (marcField.indicatorValue() != null) {
+        clause += " and lower(marc.ind%s) = '%s'".formatted(marcField.indicatorNumber(), marcField.indicatorValue());
+      }
       if (marcField.subfield() != null) {
         clause += " and marc.subfield_no = '%s'".formatted(marcField.subfield());
       }

src/test/java/org/folio/fqm/service/FqlToSqlConverterServiceTest.java

@@ -117,7 +117,8 @@ void setup() {
                 ))))
         )
       );
-    entityType = MarcFieldFactory.addSyntheticColumns(entityType, List.of("marc_245_a", "marc_245", "marc_245_ind1"), "diku");
+    entityType = MarcFieldFactory.addSyntheticColumns(entityType,
+      List.of("marc_245_a", "marc_245", "marc_245_ind1", "marc_245_ind1_7_a", "marc_245_ind1_blank_a"), "diku");
   }
 
   static Condition trueCondition = trueCondition();
@@ -1626,6 +1627,33 @@ void shouldRejectUnsupportedOperatorsOnMarcIndicator() {
     );
   }
 
+  @Test
+  void shouldGenerateConstrainedSubfieldMarcCondition() {
+    String rendered = renderMarcCondition("""
+      {"marc_245_ind1_7_a": {"$contains": "Shakespeare"}}""");
+
+    // Single EXISTS constraining field_no + fixed indicator + subfield_no on the same row, targeting value.
+    assertTrue(rendered.contains("exists (select"));
+    assertTrue(rendered.contains("marc.field_no = '245'"));
+    assertTrue(rendered.contains("lower(marc.ind1) = '7'"));
+    assertTrue(rendered.contains("marc.subfield_no = 'a'"));
+    assertTrue(rendered.contains("lower(marc.value) like"));
+    assertTrue(rendered.toLowerCase().contains("shakespeare"));
+    // Targets the value, so text operators apply (unlike indicator-only fields): %value% wrapping present.
+    assertTrue(rendered.contains("'%' ||"));
+  }
+
+  @Test
+  void shouldMapBlankForConstrainedSubfieldMarcCondition() {
+    String rendered = renderMarcCondition("""
+      {"marc_245_ind1_blank_a": {"$eq": "History"}}""");
+
+    // The blank indicator token in the field name is fixed to the stored '#'.
+    assertTrue(rendered.contains("lower(marc.ind1) = '#'"));
+    assertTrue(rendered.contains("marc.subfield_no = 'a'"));
+    assertTrue(rendered.contains("lower(marc.value) ="));
+  }
+
   @Test
   void shouldGenerateTagOnlyMarcCondition() {
     String rendered = renderMarcCondition("""

src/test/java/org/folio/fqm/utils/MarcFieldFactoryTest.java

@@ -38,13 +38,19 @@ class MarcFieldFactoryTest {
     "marc_001,      true,  control field (tag-only)",
     "marc_245_ind1, true,  indicator 1",
     "marc_245_ind2, true,  indicator 2",
+    "marc_245_ind1_7_a,     true,  constrained subfield",
+    "marc_650_ind2_7_a,     true,  constrained subfield (ind2)",
+    "marc_245_ind1_blank_a, true,  constrained subfield with blank indicator",
     "marc_24_a,     false, tag must be exactly 3 digits",
     "marc_2451_a,   false, tag must be exactly 3 digits",
     "marc_abc_a,    false, tag must be numeric",
     "marc_245_aa,   false, subfield must be a single character",
     "245_a,         false, missing marc_ prefix",
     "marc_001_a,    false, control field has no subfields",
     "marc_008_ind1, false, control field has no indicators",
+    "marc_008_ind2_7_a,     false, control field cannot use constrained subfield",
+    "marc_245_ind3_7_a,     false, indicator number must be 1 or 2",
+    "marc_245_ind1_ab_a,    false, fixed indicator value must be single char or blank",
   })
   void shouldRecognizeMarcFieldNames(String fieldName, boolean valid, String why) {
     assertEquals(valid, MarcFieldFactory.isMarcFieldName(fieldName), why);
@@ -63,8 +69,8 @@ void shouldSupportIndicatorFields() {
     MarcFieldFactory.MarcFieldName parsed = MarcFieldFactory.parse("marc_245_ind1").orElseThrow();
     assertEquals("245", parsed.tag());
     assertNull(parsed.subfield());
-    assertEquals("1", parsed.indicator());
-    assertTrue(parsed.isIndicator());
+    assertEquals("1", parsed.indicatorNumber());
+    assertTrue(parsed.isIndicatorTarget());
     assertEquals("ind1", parsed.targetColumn());
     assertEquals("MARC 245 ind1", parsed.labelAlias());
     // Indicators target the ind1/ind2 column, matched case-insensitively like other MARC values.
@@ -90,6 +96,57 @@ void shouldSupportIndicatorFields() {
     );
   }
 
+  @Test
+  void shouldSupportConstrainedSubfieldFields() {
+    MarcFieldFactory.MarcFieldName parsed = MarcFieldFactory.parse("marc_245_ind1_7_a").orElseThrow();
+    assertEquals("245", parsed.tag());
+    assertEquals("a", parsed.subfield());
+    assertEquals("1", parsed.indicatorNumber());
+    assertEquals("7", parsed.indicatorValue());
+    // The subfield value is the target; the indicator is a fixed same-row constraint (not the target).
+    assertFalse(parsed.isIndicatorTarget());
+    assertEquals("value", parsed.targetColumn());
+    assertEquals("lower(marc.value)", parsed.filterValueGetter());
+    assertEquals("MARC 245 ind1=7 $a", parsed.labelAlias());
+
+    // Synthetic column: value target (no DISTINCT), aggregated only from rows matching the fixed indicator.
+    EntityTypeColumn column = MarcFieldFactory.createSyntheticColumn(entityTypeWithMarcSupport(), "marc_245_ind1_7_a", "diku").orElseThrow();
+    assertEquals("lower(marc.value)", column.getFilterValueGetter());
+    assertTrue(column.getValueGetter().contains("jsonb_agg(marc.value)"));
+    assertFalse(column.getValueGetter().contains("DISTINCT"));
+
+    EntityType entityType = MarcFieldFactory.addSyntheticColumns(entityTypeWithMarcSupport(), List.of("marc_245_ind1_7_a"), "diku");
+    MarcQueryContext context = MarcFieldFactory.createQueryContext(entityType, "marc_245_ind1_7_a").orElseThrow();
+    // Same-row constraint: field_no + fixed indicator + subfield_no, targeting the value column.
+    assertEquals(
+      "marc.marc_id = " + MARC_RECORD_ID_GETTER + " and marc.field_no = '245' and lower(marc.ind1) = '7' and marc.subfield_no = 'a'",
+      context.whereClause()
+    );
+    assertEquals(
+      "exists (select 1 from diku_mod_fqm_manager.src_srs_marc_indexers marc where "
+        + "marc.marc_id = " + MARC_RECORD_ID_GETTER + " and marc.field_no = '245' and lower(marc.ind1) = '7' and marc.subfield_no = 'a' "
+        + "and lower(marc.value) = {0})",
+      context.existsClause("=", true)
+    );
+  }
+
+  @Test
+  void shouldMapBlankAndNormalizeCaseForConstrainedSubfield() {
+    // Blank token maps to the stored '#' in SQL, but the label shows the readable "blank".
+    MarcFieldFactory.MarcFieldName blank = MarcFieldFactory.parse("marc_245_ind1_blank_a").orElseThrow();
+    assertEquals("#", blank.indicatorValue());
+    assertTrue(blank.indicatorConstraintClause().contains("lower(marc.ind1) = '#'"));
+    assertEquals("MARC 245 ind1=blank $a", blank.labelAlias());
+
+    // Uppercase input is accepted and normalized (original field name preserved; tag/subfield/indicator normalized).
+    MarcFieldFactory.MarcFieldName upper = MarcFieldFactory.parse("MARC_245_IND1_7_A").orElseThrow();
+    assertEquals("MARC_245_IND1_7_A", upper.fieldName());
+    assertEquals("245", upper.tag());
+    assertEquals("a", upper.subfield());
+    assertEquals("1", upper.indicatorNumber());
+    assertEquals("7", upper.indicatorValue());
+  }
+
   @Test
   void shouldParseTagOnlyFieldNameWithNoSubfield() {
     MarcFieldFactory.MarcFieldName parsed = MarcFieldFactory.parse("marc_245").orElseThrow();