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
+  // Counter columns support reads and filters, but not writes through the Data API.
+  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,16 @@ public final void createDefaultTablesAndIndexes() {
       assertThat(executeCqlStatement(createTable.build())).isTrue();
     }
 
+    @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 +216,30 @@ public final void updateCounter() {
               "The operation was not supported by the columns: counter(counter).");
     }
 
-    // TODO filter on counter, INVALID_FILTER_COLUMN_VALUES
+    // Seed with CQL because the Data API does not support counter writes.
     @Test
     @Order(4)
     public final void filterCounter() {
+      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"))
+          .body("data.document.counter", is(5));
     }
   }
 

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

@@ -439,6 +439,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,9 @@ 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),
+        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),
@@ -906,6 +909,11 @@ private static Stream<Arguments> outOfRangeOfCqlNumberTestCases() {
             DataTypes.BIGINT,
             TEST_DATA.OUT_OF_RANGE_FOR_BIGINT.toBigIntegerExact(),
             "BigInteger out of long range"),
+        Arguments.of(DataTypes.COUNTER, TEST_DATA.OUT_OF_RANGE_FOR_BIGINT, "Overflow"),
+        Arguments.of(
+            DataTypes.COUNTER,
+            TEST_DATA.OUT_OF_RANGE_FOR_BIGINT.toBigIntegerExact(),
+            "BigInteger out of long range"),
         Arguments.of(DataTypes.INT, TEST_DATA.OVERFLOW_FOR_INT, "Overflow"),
         Arguments.of(
             DataTypes.INT,
@@ -949,6 +957,7 @@ private static Stream<Arguments> nonExactToCqlIntegerTestCases() {
     // Arguments: (DataType, Number-not-exact-as-integer)
     return Stream.of(
         Arguments.of(DataTypes.BIGINT, TEST_DATA.NOT_EXACT_AS_INTEGER),
+        Arguments.of(DataTypes.COUNTER, TEST_DATA.NOT_EXACT_AS_INTEGER),
         Arguments.of(DataTypes.INT, TEST_DATA.NOT_EXACT_AS_INTEGER),
         Arguments.of(DataTypes.SMALLINT, TEST_DATA.NOT_EXACT_AS_INTEGER),
         Arguments.of(DataTypes.TINYINT, TEST_DATA.NOT_EXACT_AS_INTEGER),