[CALCITE-2226] fix SubstringConverter indexes translation

druid/src/main/java/org/apache/calcite/adapter/druid/SubstringOperatorConversion.java

@@ -20,6 +20,7 @@
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 
@@ -43,20 +44,41 @@ public class SubstringOperatorConversion implements DruidSqlOperatorConverter {
       return null;
     }
 
-    final int index = RexLiteral.intValue(call.getOperands().get(1)) - 1;
+    final String startIndex;
+    final String length;
     // SQL is 1-indexed, Druid is 0-indexed.
-    final int length;
+    if (!call.getOperands().get(1).isA(SqlKind.LITERAL)) {
+      final String arg1 = DruidExpressions.toDruidExpression(
+          call.getOperands().get(1), rowType, query);
+      if (arg1 == null) {
+        // can not infer start index expression bailout.
+        return null;
+      }
+      startIndex = DruidQuery.format("(%s - 1)", arg1);
+    } else {
+      startIndex = DruidExpressions.numberLiteral(
+          RexLiteral.intValue(call.getOperands().get(1)) - 1);
+    }
+
     if (call.getOperands().size() > 2) {
-      //case substring from index with length
-      length = RexLiteral.intValue(call.getOperands().get(2));
+      //case substring from start index with length
+      if (!call.getOperands().get(2).isA(SqlKind.LITERAL)) {
+        // case it is an expression try to parse it
+        length = DruidExpressions.toDruidExpression(
+            call.getOperands().get(2), rowType, query);
+        if (length == null) {
+          return null;
+        }
+      } else {
+        // case length is a constant
+        length = DruidExpressions.numberLiteral(RexLiteral.intValue(call.getOperands().get(2)));
+      }
+
     } else {
       //case substring from index to the end
-      length = -1;
+      length = DruidExpressions.numberLiteral(-1);
     }
-    return DruidQuery.format("substring(%s, %s, %s)",
-        arg,
-        DruidExpressions.numberLiteral(index),
-        DruidExpressions.numberLiteral(length));
+    return DruidQuery.format("substring(%s, %s, %s)", arg, startIndex, length);
   }
 }
 

druid/src/test/java/org/apache/calcite/test/DruidAdapterIT.java

@@ -3888,6 +3888,39 @@ public void testSubStringExpressionFilter() {
                     + "\"outputType\":\"STRING\"}]"));
   }
 
+  @Test
+  public void testSubStringWithNonConstantIndexes() {
+    final String sql = "SELECT COUNT(*) FROM "
+        + FOODMART_TABLE
+        + " WHERE SUBSTRING(\"product_id\" from CAST(\"store_cost\" as INT)/1000 + 2  "
+        + "for CAST(\"product_id\" as INT)) like '1%'";
+
+    sql(sql, FOODMART).returnsOrdered("EXPR$0=10893")
+        .queryContains(
+            druidChecker("\"queryType\":\"timeseries\"", "like(substring(\\\"product_id\\\""))
+        .explainContains(
+            "PLAN=EnumerableInterpreter\n  DruidQuery(table=[[foodmart, foodmart]], "
+                + "intervals=[[1900-01-09T00:00:00.000Z/2992-01-10T00:00:00.000Z]], "
+                + "filter=[LIKE(SUBSTRING($1, +(/(CAST($91):INTEGER, 1000), 2), CAST($1):INTEGER), '1%')], "
+                + "groups=[{}], aggs=[[COUNT()]])\n\n");
+  }
+
+  @Test
+  public void testSubStringWithNonConstantIndex() {
+    final String sql = "SELECT COUNT(*) FROM "
+        + FOODMART_TABLE
+        + " WHERE SUBSTRING(\"product_id\" from CAST(\"store_cost\" as INT)/1000 + 1) like '1%'";
+
+    sql(sql, FOODMART).returnsOrdered("EXPR$0=36839")
+        .queryContains(druidChecker("like(substring(\\\"product_id\\\""))
+        .explainContains(
+            "PLAN=EnumerableInterpreter\n  DruidQuery(table=[[foodmart, foodmart]], "
+                + "intervals=[[1900-01-09T00:00:00.000Z/2992-01-10T00:00:00.000Z]], "
+                + "filter=[LIKE(SUBSTRING($1, +(/(CAST($91):INTEGER, 1000), 1)), '1%')],"
+                + " groups=[{}], aggs=[[COUNT()]])\n\n");
+  }
+
+
   /**
    * Test case for https://issues.apache.org/jira/browse/CALCITE-2098.
    * Need to make sure that when there we have a valid filter with no conjunction we still push