Changing the array behavior to start the index from 1.

docs/syntax-reference.rst

@@ -166,13 +166,15 @@ The supported column data types are:
 -  ``BIGINT``
 -  ``DOUBLE``
 -  ``VARCHAR`` (or ``STRING``)
--  ``ARRAY<ArrayType>`` (JSON and AVRO only. Index starts from 0)
+-  ``ARRAY<ArrayType>`` (JSON and AVRO only. Index starts from 1)
 -  ``MAP<VARCHAR, ValueType>`` (JSON and AVRO only)
 -  ``STRUCT<FieldName FieldType, ...>`` (JSON and AVRO only) The STRUCT type requires you to specify a list of fields.
    For each field you must specify the field name (FieldName) and field type (FieldType). The field type can be any of
    the supported KSQL types, including the complex types ``MAP``, ``ARRAY``, and ``STRUCT``. ``STRUCT`` fields can be
    accessed in expressions using the struct dereference (``->``) operator. See :ref:`operators` for more details.
 
+Note that starting KSQL 5.1 array index starts from 1. For the versions of KSQL before 5.1, array index starts from 0.
+
 KSQL adds the implicit columns ``ROWTIME`` and ``ROWKEY`` to every
 stream and table, which represent the corresponding Kafka message
 timestamp and message key, respectively. The timestamp has milliseconds accuracy.
@@ -243,13 +245,15 @@ The supported column data types are:
 -  ``BIGINT``
 -  ``DOUBLE``
 -  ``VARCHAR`` (or ``STRING``)
--  ``ARRAY<ArrayType>`` (JSON and AVRO only. Index starts from 0)
+-  ``ARRAY<ArrayType>`` (JSON and AVRO only. Index starts from 1)
 -  ``MAP<VARCHAR, ValueType>`` (JSON and AVRO only)
 -  ``STRUCT<FieldName FieldType, ...>`` (JSON and AVRO only) The STRUCT type requires you to specify a list of fields.
    For each field you must specify the field name (FieldName) and field type (FieldType). The field type can be any of
    the supported KSQL types, including the complex types ``MAP``, ``ARRAY``, and ``STRUCT``. ``STRUCT`` fields can be
    accessed in expressions using the struct dereference (``->``) operator. See :ref:`operators` for more details.
 
+Note that starting KSQL 5.1 array index starts from 1. For the versions of KSQL before 5.1, array index starts from 0.
+
 KSQL adds the implicit columns ``ROWTIME`` and ``ROWKEY`` to every
 stream and table, which represent the corresponding Kafka message
 timestamp and message key, respectively. The timestamp has milliseconds accuracy.
@@ -987,10 +991,11 @@ The explanation for each operator includes a supporting example based on the fol
 
 - Subscript (``[subscript_expr]``) The subscript operator is used to reference the value at
   an array index or a map key.
+  Note that starting KSQL 5.1 array index starts from 1. For the versions of KSQL before 5.1, array index starts from 0.
 
 .. code:: sql
 
-  SELECT NICKNAMES[0] FROM USERS;
+  SELECT NICKNAMES[1] FROM USERS;
 
 .. _functions:
 

ksql-cli/src/test/java/io/confluent/ksql/CliTest.java

@@ -435,7 +435,7 @@ public void testSelectLimit() {
   @Test
   public void testSelectUDFs() {
     final String selectColumns =
-        "ITEMID, ORDERUNITS*10, PRICEARRAY[0]+10, KEYVALUEMAP['key1']*KEYVALUEMAP['key2']+10, PRICEARRAY[1]>1000";
+        "ITEMID, ORDERUNITS*10, PRICEARRAY[1]+10, KEYVALUEMAP['key1']*KEYVALUEMAP['key2']+10, PRICEARRAY[1]>1000";
     final String whereClause = "ORDERUNITS > 20 AND ITEMID LIKE '%_8'";
 
     final String queryString = String.format(

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

@@ -532,7 +532,7 @@ protected Pair<String, Schema> visitSubscriptExpression(
       switch (internalSchema.type()) {
         case ARRAY:
           return new Pair<>(
-              String.format("((%s) ((%s)%s).get((int)(%s)))",
+              String.format("((%s) ((%s)%s).get(((int)(%s)) - 1))",
                   SchemaUtil.getJavaType(internalSchema.valueSchema()).getSimpleName(),
                   internalSchemaJavaType,
                   process(node.getBase(), unmangleNames).getLeft(),

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

@@ -183,7 +183,7 @@ public void testIsNull() throws Exception {
 
     @Test
     public void shouldHandleMultiDimensionalArray() throws Exception {
-        final String simpleQuery = "SELECT col14[0][0] FROM CODEGEN_TEST;";
+        final String simpleQuery = "SELECT col14[1][1] FROM CODEGEN_TEST;";
         final Analysis analysis = analyzeQuery(simpleQuery);
         final ExpressionMetadata expressionEvaluatorMetadata = codeGenRunner.buildCodeGenFromParseTree
             (analysis.getSelectExpressions().get(0));

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

@@ -73,14 +73,14 @@ public void shouldProcessBasicJavaMath() {
 
   @Test
   public void shouldProcessArrayExpressionCorrectly() {
-    String simpleQuery = "SELECT col4[0] FROM test1 WHERE col0 > 100;";
+    String simpleQuery = "SELECT col4[1] FROM test1 WHERE col0 > 100;";
     Analysis analysis = analyzeQuery(simpleQuery);
 
     String javaExpression = new SqlToJavaVisitor(schema, functionRegistry)
         .process(analysis.getSelectExpressions().get(0));
 
     assertThat(javaExpression,
-        equalTo("((Double) ((java.util.List)TEST1_COL4).get((int)(Integer.parseInt(\"0\"))))"));
+        equalTo("((Double) ((java.util.List)TEST1_COL4).get(((int)(Integer.parseInt(\"1\"))) - 1))"));
   }
 
   @Test

ksql-engine/src/test/java/io/confluent/ksql/integration/UdfIntTest.java

@@ -143,7 +143,7 @@ private void testApplyUdfsToColumns(String resultStreamName,
     final String queryString = String.format(
         "CREATE STREAM %s AS SELECT %s FROM %s WHERE %s;",
         resultStreamName,
-        "ITEMID, ORDERUNITS*10, PRICEARRAY[0]+10, KEYVALUEMAP['key1']*KEYVALUEMAP['key2']+10, "
+        "ITEMID, ORDERUNITS*10, PRICEARRAY[1]+10, KEYVALUEMAP['key1']*KEYVALUEMAP['key2']+10, "
         + "PRICEARRAY[1]>1000",
         inputStreamName,
         "ORDERUNITS > 20 AND ITEMID LIKE '%_8'"

ksql-engine/src/test/java/io/confluent/ksql/util/ExpressionTypeManagerTest.java

@@ -179,7 +179,7 @@ public void shouldFailIfThereIsInvalidFieldNameInStructCall() {
 
   @Test
   public void shouldFindTheNestedArrayTypeCorrectly() {
-    final Analysis analysis = analyzeQuery("SELECT ARRAYCOL[0]->CATEGORY->NAME, NESTED_ORDER_COL->arraycol[0] from NESTED_STREAM;");
+    final Analysis analysis = analyzeQuery("SELECT ARRAYCOL[1]->CATEGORY->NAME, NESTED_ORDER_COL->arraycol[1] from NESTED_STREAM;");
     final ExpressionTypeManager expressionTypeManager = new ExpressionTypeManager(metaStore.getSource("NESTED_STREAM").getSchema(),
         functionRegistry);
     assertThat(expressionTypeManager.getExpressionSchema(analysis.getSelectExpressions().get(0)),

ksql-engine/src/test/resources/query-validation-tests/extract-json-field.json

@@ -22,7 +22,7 @@
       "name": "extract JSON array field",
       "statements": [
         "CREATE STREAM TEST (array_field ARRAY<VARCHAR>) WITH (kafka_topic='test_topic', value_format='JSON');",
-        "CREATE STREAM OUTPUT AS SELECT EXTRACTJSONFIELD(array_field[0], '$.nested') AS Col1, EXTRACTJSONFIELD(array_field[1], '$.nested') AS Col2 FROM TEST;"
+        "CREATE STREAM OUTPUT AS SELECT EXTRACTJSONFIELD(array_field[1], '$.nested') AS Col1, EXTRACTJSONFIELD(array_field[2], '$.nested') AS Col2 FROM TEST;"
       ],
       "inputs": [
         {"topic": "test_topic", "key": 0, "value": {"array_field": ["{\"nested\": \"nest0\"}","{\"nested\": \"nest1\"}"]}, "timestamp": 0},

ksql-engine/src/test/resources/query-validation-tests/more-complex-struct.json

@@ -800,7 +800,7 @@
       "name": "complex struct select array and map items",
       "statements": [
         "CREATE STREAM foo (NESTED_STRUCT_FIELD STRUCT <SL1_STRUCT  STRUCT< SL2_STRUCT STRUCT<SL3_STRUCT STRUCT< SL4_STRING VARCHAR, SL4_INT INT, SL4_DOUBLE_ARRAY ARRAY<DOUBLE>>, SL3_DOUBLE_ARRAY ARRAY<DOUBLE>, SL3_STRING VARCHAR>, SL2_STRUCT_MAP MAP<VARCHAR, STRUCT<SL2_3_STRING VARCHAR, SL2_3_DOUBLE DOUBLE>>, SL2_DOUBLE DOUBLE, SL2_BOOLEAN BOOLEAN >, SL1_STRING VARCHAR, SL1_STRUCT_ARRAY ARRAY<STRUCT<SL1_2_STRING VARCHAR, SL1_2_DOUBLE DOUBLE>> > , COL1 BIGINT, COL2 VARCHAR) WITH (kafka_topic='test_topic', value_format='json');",
-        "CREATE STREAM s3 AS SELECT NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT->SL3_STRUCT->SL4_DOUBLE_ARRAY[0] AS COL1, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT_MAP['mapkey'] AS COL2, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT->SL3_STRUCT->SL4_DOUBLE_ARRAY[10-5-5] AS COL3, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT_MAP['mapkey']->SL2_3_DOUBLE AS COL4 from foo;"
+        "CREATE STREAM s3 AS SELECT NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT->SL3_STRUCT->SL4_DOUBLE_ARRAY[1] AS COL1, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT_MAP['mapkey'] AS COL2, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT->SL3_STRUCT->SL4_DOUBLE_ARRAY[10-5-5+1] AS COL3, NESTED_STRUCT_FIELD->SL1_STRUCT->SL2_STRUCT_MAP['mapkey']->SL2_3_DOUBLE AS COL4 from foo;"
       ],
       "inputs": [
         {

ksql-engine/src/test/resources/query-validation-tests/project-filter.json

@@ -99,7 +99,7 @@
       "name": "Json Multi Dimensional Array",
       "statements": [
         "CREATE STREAM array_array (ID BIGINT, ARRAY_COL ARRAY<ARRAY<VARCHAR>>) WITH (kafka_topic='test_topic', value_format='JSON');",
-        "CREATE STREAM S3 as SELECT ID, ARRAY_COL[0][1] AS array_item FROM array_array;"
+        "CREATE STREAM S3 as SELECT ID, ARRAY_COL[1][2] AS array_item FROM array_array;"
       ],
       "inputs": [
         {"topic": "test_topic", "key": 0, "timestamp": 0, "value": {"id": 1, "array_col": [["item_00_1","item_01_1"],["item_10_1","item_11_1"]]}},
@@ -120,7 +120,7 @@
       "name": "Json Multi Dimensional Array",
       "statements": [
         "CREATE STREAM array_array (ID BIGINT, ARRAY_COL ARRAY<ARRAY<ARRAY<VARCHAR>>>) WITH (kafka_topic='test_topic', value_format='JSON');",
-        "CREATE STREAM S3 as SELECT ID, ARRAY_COL[0][0][0] AS array_item FROM array_array;"
+        "CREATE STREAM S3 as SELECT ID, ARRAY_COL[1][1][1] AS array_item FROM array_array;"
       ],
       "inputs": [
         {"topic": "test_topic", "key": 0, "timestamp": 0, "value": {"id": 1, "array_col": [[["item_000_1","item_001_1"],["item_010_1","item_011_1"]],[["item_100_1","item_101_1"],["item_110_1","item_111_1"]]]}},

ksql-parser/src/test/java/io/confluent/ksql/parser/KsqlParserTest.java

@@ -185,7 +185,7 @@ public void testProjection() {
 
   @Test
   public void testProjectionWithArrayMap() {
-    String queryStr = "SELECT col0, col2, col3, col4[0], col5['key1'] FROM test1;";
+    String queryStr = "SELECT col0, col2, col3, col4[1], col5['key1'] FROM test1;";
     Statement statement = KSQL_PARSER.buildAst(queryStr, metaStore).get(0);
     Assert.assertTrue("testProjectionWithArrayMap fails", statement instanceof Query);
     Query query = (Query) statement;
@@ -201,7 +201,7 @@ public void testProjectionWithArrayMap() {
     SingleColumn column3 = (SingleColumn)querySpecification.getSelect().getSelectItems().get(3);
     SingleColumn column4 = (SingleColumn)querySpecification.getSelect().getSelectItems().get(4);
     Assert.assertTrue("testProjectionWithArrayMap fails", column3.getExpression().toString()
-        .equalsIgnoreCase("TEST1.COL4[0]"));
+        .equalsIgnoreCase("TEST1.COL4[1]"));
     Assert.assertTrue("testProjectionWithArrayMap fails", column4.getExpression().toString()
         .equalsIgnoreCase("TEST1.COL5['key1']"));
   }

ksql-parser/src/test/java/io/confluent/ksql/parser/rewrite/StatementRewriteForStructTest.java

@@ -88,7 +88,7 @@ public void shouldNotCreateFunctionCallIfNotNeeded() {
 
   @Test
   public void shouldCreateCorrectFunctionCallExpressionWithSubscript() {
-    final String simpleQuery = "SELECT arraycol[0]->name as n0, mapcol['key']->name as n1 FROM nested_stream;";
+    final String simpleQuery = "SELECT arraycol[1]->name as n0, mapcol['key']->name as n1 FROM nested_stream;";
     final Statement statement = KSQL_PARSER.buildAst(simpleQuery, metaStore).get(0);
 
     final QuerySpecification querySpecification = getQuerySpecification(statement);
@@ -102,7 +102,7 @@ public void shouldCreateCorrectFunctionCallExpressionWithSubscript() {
     assertThat(col1, instanceOf(FunctionCall.class));
 
     assertThat(col0.toString(),
-        equalTo("FETCH_FIELD_FROM_STRUCT(NESTED_STREAM.ARRAYCOL[0], 'NAME')"));
+        equalTo("FETCH_FIELD_FROM_STRUCT(NESTED_STREAM.ARRAYCOL[1], 'NAME')"));
     assertThat(col1.toString(),
         equalTo("FETCH_FIELD_FROM_STRUCT(NESTED_STREAM.MAPCOL['key'], 'NAME')"));
   }

ksql-parser/src/test/java/io/confluent/ksql/parser/rewrite/StatementRewriterTest.java

@@ -78,7 +78,7 @@ public void testProjection() {
 
   @Test
   public void testProjectionWithArrayMap() {
-    final String queryStr = "SELECT col0, col2, col3, col4[0], col5['key1'] FROM test1;";
+    final String queryStr = "SELECT col0, col2, col3, col4[1], col5['key1'] FROM test1;";
     final Statement statement = KSQL_PARSER.buildAst(queryStr, metaStore).get(0);
 
     final StatementRewriter statementRewriter = new StatementRewriter();
@@ -98,7 +98,7 @@ public void testProjectionWithArrayMap() {
 
     final SingleColumn column3 = (SingleColumn)querySpecification.getSelect().getSelectItems().get(3);
     final SingleColumn column4 = (SingleColumn)querySpecification.getSelect().getSelectItems().get(4);
-    assertThat("testProjectionWithArrayMap fails", column3.getExpression().toString(), equalTo("TEST1.COL4[0]"));
+    assertThat("testProjectionWithArrayMap fails", column3.getExpression().toString(), equalTo("TEST1.COL4[1]"));
     assertThat("testProjectionWithArrayMap fails", column4.getExpression().toString(), equalTo("TEST1.COL5['key1']"));
   }