Fix #2462: add api support for filtering counter columns

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/table/codecs/JSONCodecRegistries.java

@@ -28,6 +28,8 @@ private JSONCodecRegistries() {}
                 JSONCodecs.BIGINT_FROM_BIG_DECIMAL,
                 JSONCodecs.BIGINT_FROM_BIG_INTEGER,
                 JSONCodecs.BIGINT_FROM_LONG,
+                JSONCodecs.COUNTER_FROM_BIG_DECIMAL,
+                JSONCodecs.COUNTER_FROM_BIG_INTEGER,
                 JSONCodecs.COUNTER_FROM_LONG,
                 JSONCodecs.INT_FROM_BIG_DECIMAL,
                 JSONCodecs.INT_FROM_BIG_INTEGER,

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/table/codecs/JSONCodecs.java

@@ -60,7 +60,21 @@ public abstract class JSONCodecs {
           JSONCodec.ToCQL.unsafeIdentity(),
           JSONCodec.ToJSON.unsafeNodeFactory(JsonNodeFactory.instance::numberNode));
 
-  // we can only read counters from CQL, do not support writing them
+  // Counters are read/filter-only; mirror BIGINT's variants so filters can bind them (#2462).
+  public static final JSONCodec<BigDecimal, Long> COUNTER_FROM_BIG_DECIMAL =
+      new JSONCodec<>(
+          GenericType.BIG_DECIMAL,
+          DataTypes.COUNTER,
+          JSONCodec.ToCQL.safeNumber(BigDecimal::longValueExact),
+          JSONCodec.ToJSON.unsafeNodeFactory(JsonNodeFactory.instance::numberNode));
+
+  public static final JSONCodec<BigInteger, Long> COUNTER_FROM_BIG_INTEGER =
+      new JSONCodec<>(
+          GenericType.BIG_INTEGER,
+          DataTypes.COUNTER,
+          JSONCodec.ToCQL.safeNumber(BigInteger::longValueExact),
+          JSONCodec.ToJSON.unsafeNodeFactory(JsonNodeFactory.instance::numberNode));
+
   public static final JSONCodec<Long, Long> COUNTER_FROM_LONG =
       new JSONCodec<>(
           GenericType.LONG,

src/test/java/io/stargate/sgv2/jsonapi/api/v1/tables/UnsupportedTypeTableIntegrationTest.java

@@ -161,6 +161,19 @@ public final void createDefaultTablesAndIndexes() {
       assertThat(executeCqlStatement(createTable.build())).isTrue();
     }
 
+    // Regression test for #2462: a counter column must report apiSupport.filter == true in the
+    // table description, because counters ARE filterable via ALLOW FILTERING (they just cannot be
+    // indexed). Filtering succeeds with a MISSING_INDEX warning (see filterCounter() below).
+    @Test
+    @Order(2)
+    public final void listTablesReportsCounterFilterable() {
+      assertNamespaceCommand(keyspaceName)
+          .templated()
+          .listTables(true)
+          .wasSuccessful()
+          .hasTableColumnApiSupport(TABLE_COUNTER, "counter", false, false, true, true);
+    }
+
     // In Cassandra, you cannot use an INSERT statement directly for counters.
     // Instead, counter columns require a special kind of update operation.
     // Counters in Cassandra are designed to increment or decrement a value,
@@ -206,17 +219,33 @@ public final void updateCounter() {
               "The operation was not supported by the columns: counter(counter).");
     }
 
-    // TODO filter on counter, INVALID_FILTER_COLUMN_VALUES
+    // #2462: counters are filterable via ALLOW FILTERING. Counters cannot be inserted through the
+    // API, so seed a value with raw CQL (UPDATE ... counter = counter + N), then confirm the filter
+    // binds the value, returns the row, and emits a MISSING_INDEX warning (counters can't be
+    // indexed).
     @Test
     @Order(4)
     public final void filterCounter() {
+      // keyspace/table are case-sensitive identifiers, so quote them in the raw CQL.
+      var ksCql =
+          CqlIdentifierUtil.cqlIdentifierToCQL(
+              CqlIdentifierUtil.cqlIdentifierFromUserInput(keyspaceName));
+      var tableCql =
+          CqlIdentifierUtil.cqlIdentifierToCQL(
+              CqlIdentifierUtil.cqlIdentifierFromUserInput(TABLE_COUNTER));
+      assertThat(
+              executeCqlStatement(
+                  SimpleStatement.newInstance(
+                      "UPDATE %s.%s SET counter = counter + 5 WHERE id = '1'"
+                          .formatted(ksCql, tableCql))))
+          .isTrue();
+
       assertTableCommand(keyspaceName, TABLE_COUNTER)
           .templated()
-          .findOne(ImmutableMap.of("counter", 1), null)
-          .hasSingleApiError(
-              FilterException.Code.INVALID_FILTER_COLUMN_VALUES,
-              FilterException.class,
-              "Only values that are supported by the column data type can be included");
+          .findOne(ImmutableMap.of("counter", 5), null)
+          .hasNoErrors()
+          .hasSingleWarning(WarningException.Code.MISSING_INDEX)
+          .body("data.document.id", is("1"));
     }
   }
 

src/test/java/io/stargate/sgv2/jsonapi/api/v1/util/DataApiResponseValidator.java

@@ -434,6 +434,26 @@ public DataApiResponseValidator doesNotHaveIndexes(String... indexes) {
     return toReturn;
   }
 
+  /**
+   * Asserts the {@code apiSupport} flags reported for a column in a {@code listTables} (with
+   * explain) response, e.g. {@code status.tables[name].definition.columns[column].apiSupport}.
+   */
+  public DataApiResponseValidator hasTableColumnApiSupport(
+      String tableName,
+      String columnName,
+      boolean createTable,
+      boolean insert,
+      boolean read,
+      boolean filter) {
+    var apiSupportPath =
+        "status.tables.find { it.name == '%s' }.definition.columns.%s.apiSupport"
+            .formatted(tableName, columnName);
+    return body(apiSupportPath + ".createTable", is(createTable))
+        .body(apiSupportPath + ".insert", is(insert))
+        .body(apiSupportPath + ".read", is(read))
+        .body(apiSupportPath + ".filter", is(filter));
+  }
+
   public DataApiResponseValidator hasNextPageState() {
     return body("data.nextPageState", is(notNullValue()));
   }

src/test/java/io/stargate/sgv2/jsonapi/service/operation/filters/table/codecs/JSONCodecRegistryTest.java

@@ -204,6 +204,10 @@ private static Stream<Arguments> validCodecToCQLTestCasesInt() {
         Arguments.of(DataTypes.BIGINT, -456L, -456L),
         Arguments.of(DataTypes.BIGINT, BigInteger.valueOf(123), 123L),
         Arguments.of(DataTypes.BIGINT, BigDecimal.valueOf(999.0), 999L),
+        // counters are filterable (#2462): they bind the same 3 Java types as BIGINT
+        Arguments.of(DataTypes.COUNTER, -456L, -456L),
+        Arguments.of(DataTypes.COUNTER, BigInteger.valueOf(123), 123L),
+        Arguments.of(DataTypes.COUNTER, BigDecimal.valueOf(999.0), 999L),
         Arguments.of(DataTypes.INT, -42000L, -42000),
         Arguments.of(DataTypes.INT, BigInteger.valueOf(19000), 19000),
         Arguments.of(DataTypes.INT, BigDecimal.valueOf(23456.0), 23456),