feat: Arrays should be indexable from the end too (#3004)

* Fixes #2974 Arrays should be indexable from the end too

docs/developer-guide/query-with-arrays-and-maps.rst

@@ -104,6 +104,35 @@ Your output should resemble:
 
 Press Ctrl+C to terminate the query.
 
+You can access array elements by using positive or negative index values.
+For example, to get the user's last interest run the following SELECT statement:
+
+.. code:: sql
+
+      SELECT interests[-1] AS last_interest,
+             userid,
+             gender,
+             regionid
+      FROM users_extended;
+
+Your output should resemble:  
+
+::
+
+    Travel | User_9 | OTHER  | Region_6
+    Travel | User_2 | FEMALE | Region_5
+    Sport  | User_3 | FEMALE | Region_8
+    Movies | User_5 | OTHER  | Region_9
+    Movies | User_8 | MALE   | Region_1
+    Movies | User_1 | MALE   | Region_6
+    News   | User_4 | MALE   | Region_9
+    Movies | User_7 | OTHER  | Region_1
+    Sport  | User_6 | FEMALE | Region_5
+    ^CQuery terminated
+
+
+Press Ctrl+C to terminate the query.
+
 Next Steps
 **********
 

ksql-engine/src/main/java/io/confluent/ksql/codegen/SqlToJavaVisitor.java

@@ -652,15 +652,20 @@ protected Pair<String, Schema> visitSubscriptExpression(
           SchemaUtil.getJavaType(internalSchema).getCanonicalName();
       switch (internalSchema.type()) {
         case ARRAY:
-          return new Pair<>(
-              String.format("((%s) ((%s)%s).get((int)(%s)))",
-                  SchemaUtil.getJavaType(internalSchema.valueSchema()).getSimpleName(),
-                  internalSchemaJavaType,
-                  process(node.getBase(), context).getLeft(),
-                  process(node.getIndex(), context).getLeft()
-              ),
-              internalSchema.valueSchema()
-          );
+          final String listName = process(node.getBase(), context).getLeft();
+          final String suppliedIdx = process(node.getIndex(), context).getLeft();
+          final String trueIdx = node.getIndex().toString().startsWith("-")
+              ? String.format("((%s)%s).size()%s", internalSchemaJavaType, listName, suppliedIdx)
+              : suppliedIdx;
+
+          final String code = format("((%s) ((%s)%s).get((int)%s))",
+              SchemaUtil.getJavaType(internalSchema.valueSchema()).getSimpleName(),
+              internalSchemaJavaType,
+              listName,
+              trueIdx);
+
+          return new Pair<>(code, internalSchema.valueSchema());
+
         case MAP:
           return new Pair<>(
               String.format("((%s) ((%s)%s).get(%s))",
@@ -947,4 +952,4 @@ private CaseWhenProcessed(
     }
   }
 
-}
+}

ksql-engine/src/test/java/io/confluent/ksql/codegen/SqlToJavaVisitorTest.java

@@ -103,7 +103,19 @@ public void shouldProcessArrayExpressionCorrectly() {
         .process(analysis.getSelectExpressions().get(0));
 
     assertThat(javaExpression,
-        equalTo("((Double) ((java.util.List)TEST1_COL4).get((int)(0)))"));
+       equalTo("((Double) ((java.util.List)TEST1_COL4).get((int)0))"));
+  }
+
+  @Test
+  public void shouldProcessArrayNegativeIndexExpressionCorrectly() {
+    final String simpleQuery = "SELECT col4[-1] FROM test1;";
+    final Analysis analysis = analyzeQuery(simpleQuery, metaStore);
+
+    final String javaExpression = sqlToJavaVisitor
+            .process(analysis.getSelectExpressions().get(0));
+
+    assertThat(javaExpression,
+            equalTo("((Double) ((java.util.List)TEST1_COL4).get((int)((java.util.List)TEST1_COL4).size()-1))"));
   }
 
   @Test
@@ -650,4 +662,4 @@ public void shouldThrowOnNotIn() {
     // When:
     sqlToJavaVisitor.process(decimal);
   }
-}
+}

ksql-functional-tests/src/test/resources/query-validation-tests/arrayindex.json

@@ -0,0 +1,51 @@
+{
+  "comments": [
+    "Arrays should be indexable from the end too",
+    "https://github.com/confluentinc/ksql/issues/2974"
+  ],
+  "tests": [
+
+    {
+      "name": "select the first element of an Array",
+      "statements": [
+        "CREATE STREAM test (colors ARRAY<STRING>) WITH (kafka_topic='test_topic', value_format='JSON');",
+        "CREATE STREAM OUTPUT AS SELECT colors[0] as C FROM test;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "value": {"colors": ["Red", "Green"]}},
+        {"topic": "test_topic", "value": {"colors": ["Black"]}},
+        {"topic": "test_topic", "value": {"colors": [null, "Yellow", "Pink"]}},
+        {"topic": "test_topic", "value": {"colors": []}}
+      ],
+      "outputs": [
+        {"topic": "OUTPUT", "value": {"C": "Red"}},
+        {"topic": "OUTPUT", "value": {"C": "Black"}},
+        {"topic": "OUTPUT", "value": {"C": null}},
+        {"topic": "OUTPUT", "value": {"C": null}}
+      ]
+    },
+    {
+      "name": "select the last element of an Array (-1)",
+      "statements": [
+        "CREATE STREAM test (colors ARRAY<STRING>) WITH (kafka_topic='test_topic', value_format='JSON');",
+        "CREATE STREAM OUTPUT AS SELECT colors[-1] as C FROM test;"
+      ],
+      "inputs": [
+        {"topic": "test_topic", "value": {"colors": ["Red", "Green"]}},
+        {"topic": "test_topic", "value": {"colors": ["Black"]}},
+        {"topic": "test_topic", "value": {"colors": ["Pink", "Yellow", "Pink"]}},
+        {"topic": "test_topic", "value": {"colors": ["White", "Pink"]}},
+        {"topic": "test_topic", "value": {"colors": ["Pink", null]}},
+        {"topic": "test_topic", "value": {"colors": []}}
+      ],
+      "outputs": [
+        {"topic": "OUTPUT", "value": {"C": "Green"}},
+        {"topic": "OUTPUT", "value": {"C": "Black"}},
+        {"topic": "OUTPUT", "value": {"C": "Pink"}},
+        {"topic": "OUTPUT", "value": {"C": "Pink"}},
+        {"topic": "OUTPUT", "value": {"C": null}},
+        {"topic": "OUTPUT", "value": {"C": null}}
+      ]
+    }
+  ]
+}