Merge 'forward_service: fix forgetting case-sensitivity in aggregates…

… ' from Jan Ciołek

There was a bug that caused aggregates to fail when used on column-sensitive columns.

For example:
```cql
SELECT SUM("SomeColumn") FROM ks.table;
```
would fail, with a message saying that there is no column "somecolumn".

This is because the case-sensitivity got lost on the way.

For non case-sensitive column names we convert them to lowercase, but for case sensitive names we have to preserve the name as originally written.

The problem was in `forward_service` - we took a column name and created a non case-sensitive `column_identifier` out of it.
This converted the name to lowercase, and later such column couldn't be found.

To fix it, let's make the `column_identifier` case-sensitive.
It will preserve the name, without converting it to lowercase.

Fixes: #14307

Closes #14340

* github.com:scylladb/scylladb:
  service/forward_service.cc: make case-sensitivity explicit
  cql-pytest/test_aggregate: test case-sensitive column name in aggregate
  forward_service: fix forgetting case-sensitivity in aggregates

service/forward_service.cc

@@ -334,8 +334,9 @@ static shared_ptr<cql3::selection::selection> mock_selection(
         const std::optional<query::forward_request::aggregation_info>& info
     ) {
         auto name_as_expression = [] (const sstring& name) -> cql3::expr::expression {
+            constexpr bool keep_case = true;
             return cql3::expr::unresolved_identifier {
-                make_shared<cql3::column_identifier_raw>(name, false)
+                make_shared<cql3::column_identifier_raw>(name, keep_case)
             };
         };
 

test/cql-pytest/test_aggregate.py

@@ -18,6 +18,11 @@ def table1(cql, test_keyspace):
     with new_test_table(cql, test_keyspace, "p int, c int, v int, d double, dc decimal, PRIMARY KEY (p, c)") as table:
         yield table
 
+@pytest.fixture(scope="module")
+def table2(cql, test_keyspace):
+    with new_test_table(cql, test_keyspace, 'p int, c int, somecolumn int, "SomeColumn" int, "OtherColumn" int, PRIMARY KEY (p, c)') as table:
+        yield table
+
 # When there is no row matching the selection, the count should be 0.
 # First check a "=" expression matching no row:
 def test_count_empty_eq(cql, table1):
@@ -217,3 +222,22 @@ def test_avg_decimal_2(cql, table1, cassandra_bug):
 def test_reject_aggregates_in_where_clause(cql, table1):
     assert_invalid_message(cql, table1, 'Aggregation',
                            f'SELECT * FROM {table1} WHERE p = sum((int)4)')
+
+# Reproduces #13265
+# Aggregates must handle case-sensitive column names correctly.
+def test_sum_case_sensitive_column(cql, table2):
+    p = unique_key_int()
+    cql.execute(f'insert into {table2} (p, c, somecolumn, "SomeColumn", "OtherColumn") VALUES ({p}, 1, 1, 10, 100)')
+    cql.execute(f'insert into {table2} (p, c, somecolumn, "SomeColumn", "OtherColumn") VALUES ({p}, 2, 2, 20, 200)')
+    cql.execute(f'insert into {table2} (p, c, somecolumn, "SomeColumn", "OtherColumn") VALUES ({p}, 3, 3, 30, 300)')
+
+    assert cql.execute(f'select sum(somecolumn) from {table2} where p = {p}').one()[0] == 6
+    assert cql.execute(f'select sum("somecolumn") from {table2} where p = {p}').one()[0] == 6
+    assert cql.execute(f'select sum("SomeColumn") from {table2} where p = {p}').one()[0] == 60
+    assert cql.execute(f'select sum(SomeColumn) from {table2} where p = {p}').one()[0] == 6
+    assert cql.execute(f'select sum("OtherColumn") from {table2} where p = {p}').one()[0] == 600
+
+    assert_invalid_message(cql, table2, 'someColumn',
+                           f'select sum("someColumn") from {table2} where p = {p}')
+    assert_invalid_message(cql, table2, 'othercolumn',
+                           f'select sum(OtherColumn) from {table2} where p = {p}')