chore: split key format feature flag

confluentinc · Dec 2, 2020 · 47a9c95 · 47a9c95
1 parent 07dc0c7
commit 47a9c95
Show file tree

Hide file tree

Showing 16 changed files with 122 additions and 93 deletions.
diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java b/ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java
@@ -138,7 +138,17 @@ public class KsqlConfig extends AbstractConfig {
   public static final String KSQL_KEY_FORMAT_ENABLED = "ksql.key.format.enabled";
   public static final Boolean KSQL_KEY_FORMAT_ENABLED_DEFAULT = false;
   public static final String KSQL_KEY_FORMAT_ENABLED_DOC =
-      "Feature flag for non-Kafka key formats";
+      "Feature flag for key formats under development";
+
+  public static final String KSQL_COMPLEX_KEY_FORMAT_ENABLED = "ksql.complex.key.format.enabled";
+  public static final Boolean KSQL_COMPLEX_KEY_FORMAT_ENABLED_DEFAULT = false;
+  public static final String KSQL_COMPLEX_KEY_FORMAT_ENABLED_DOC =
+      "Feature flag for complex (non-primitive) keys";
+
+  public static final String KSQL_MULTICOL_KEY_FORMAT_ENABLED = "ksql.multicol.key.format.enabled";
+  public static final Boolean KSQL_MULTICOL_KEY_FORMAT_ENABLED_DEFAULT = false;
+  public static final String KSQL_MULTICOL_KEY_FORMAT_ENABLED_DOC =
+      "Feature flag for multi-column keys";
 
   public static final String KSQL_DEFAULT_KEY_FORMAT_CONFIG = "ksql.persistence.default.format.key";
   private static final String KSQL_DEFAULT_KEY_FORMAT_DEFAULT = "KAFKA";
@@ -613,6 +623,18 @@ private static ConfigDef buildConfigDef(final ConfigGeneration generation) {
             KSQL_KEY_FORMAT_ENABLED_DEFAULT,
             ConfigDef.Importance.LOW,
             KSQL_KEY_FORMAT_ENABLED_DOC
+        ).define(
+            KSQL_COMPLEX_KEY_FORMAT_ENABLED,
+            Type.BOOLEAN,
+            KSQL_COMPLEX_KEY_FORMAT_ENABLED_DEFAULT,
+            ConfigDef.Importance.LOW,
+            KSQL_COMPLEX_KEY_FORMAT_ENABLED_DOC
+        ).define(
+            KSQL_MULTICOL_KEY_FORMAT_ENABLED,
+            Type.BOOLEAN,
+            KSQL_MULTICOL_KEY_FORMAT_ENABLED_DEFAULT,
+            ConfigDef.Importance.LOW,
+            KSQL_MULTICOL_KEY_FORMAT_ENABLED_DOC
         ).define(
             KSQL_DEFAULT_KEY_FORMAT_CONFIG,
             Type.STRING,

diff --git a/ksqldb-functional-tests/src/test/java/io/confluent/ksql/test/planned/PlannedTestUtils.java b/ksqldb-functional-tests/src/test/java/io/confluent/ksql/test/planned/PlannedTestUtils.java
@@ -32,6 +32,7 @@
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 public final class PlannedTestUtils {
@@ -48,9 +49,10 @@ public static boolean isPlannedTestCase(final TestCase testCase) {
 
   public static boolean isNotExcluded(final TestCase testCase) {
     // Place temporary logic here to exclude test cases based on feature flags, etc.
-    return !(boolean) testCase
-        .properties()
-        .getOrDefault(KsqlConfig.KSQL_KEY_FORMAT_ENABLED, false);
+    final Map<String, Object> props = testCase.properties();
+    return !(boolean) props.getOrDefault(KsqlConfig.KSQL_KEY_FORMAT_ENABLED, false)
+        && !(boolean) props.getOrDefault(KsqlConfig.KSQL_COMPLEX_KEY_FORMAT_ENABLED, false)
+        && !(boolean) props.getOrDefault(KsqlConfig.KSQL_MULTICOL_KEY_FORMAT_ENABLED, false);
   }
 
   public static boolean isSamePlan(

diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/as_value.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/as_value.json
@@ -153,7 +153,7 @@
     {
       "name": "ARRAY",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (ID ARRAY<VARCHAR> KEY, V0 INT, V1 INT) WITH (kafka_topic='input', format='JSON');",
@@ -174,7 +174,7 @@
     {
       "name": "STRUCT",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (ID STRUCT<F1 VARCHAR> KEY, V0 INT, V1 INT) WITH (kafka_topic='input', format='JSON');",

diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/avro.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/avro.json
@@ -860,7 +860,7 @@
     {
       "name": "ARRAY - key - no inference",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K ARRAY<VARCHAR> KEY, foo INT) WITH (kafka_topic='input_topic', format='AVRO');",
@@ -880,7 +880,7 @@
     {
       "name": "ARRAY - key - inference",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (foo INT) WITH (kafka_topic='input_topic', format='AVRO');",
@@ -908,7 +908,7 @@
     {
       "name": "MAP - key - C*",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K MAP<INT, DOUBLE> KEY, foo INT) WITH (kafka_topic='input_topic', format='AVRO');"
@@ -921,7 +921,7 @@
     {
       "name": "nested MAP type - key - C*",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K ARRAY<MAP<INT, DOUBLE>> KEY, foo INT) WITH (kafka_topic='input_topic', format='AVRO');"
@@ -934,7 +934,7 @@
     {
       "name": "MAP - key - C*AS",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (k STRING KEY, v STRING) WITH (kafka_topic='input_topic', format='AVRO');",
@@ -948,7 +948,7 @@
     {
       "name": "nested MAP type - key - C*AS",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (k STRING KEY, v STRING) WITH (kafka_topic='input_topic', format='AVRO');",
@@ -962,7 +962,7 @@
     {
       "name": "STRUCT - key - no inference",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K STRUCT<F1 VARCHAR> KEY, foo INT) WITH (kafka_topic='input_topic', format='AVRO');",
@@ -982,7 +982,7 @@
     {
       "name": "STRUCT - key - inference",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "topics": [
         {

diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/delimited.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/delimited.json
@@ -140,7 +140,7 @@
     {
       "name": "ARRAY - C* - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K ARRAY<INT> KEY, foo INT) WITH (kafka_topic='input_topic', format='DELIMITED');"
@@ -153,7 +153,7 @@
     {
       "name": "ARRAY - C*AS - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (v INT) WITH (kafka_topic='input_topic', format='DELIMITED');",
@@ -167,7 +167,7 @@
     {
       "name": "MAP - C* - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K MAP<INT, DOUBLE> KEY, foo INT) WITH (kafka_topic='input_topic', format='DELIMITED');"
@@ -180,7 +180,7 @@
     {
       "name": "MAP - C*AS - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (k STRING KEY, v STRING) WITH (kafka_topic='input_topic', format='DELIMITED');",
@@ -194,7 +194,7 @@
     {
       "name": "STRUCT - C* - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K STRUCT<F1 DOUBLE> KEY, foo INT) WITH (kafka_topic='input_topic', format='DELIMITED');"
@@ -207,7 +207,7 @@
     {
       "name": "STRUCT - C*AS - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (v DOUBLE) WITH (kafka_topic='input_topic', format='DELIMITED');",

diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/group-by.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/group-by.json
@@ -2258,7 +2258,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "windowed aggregate with struct key",
       "format": ["JSON", "AVRO"],
@@ -2279,7 +2279,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "windowed aggregate with field within struct key",
       "statements": [
@@ -2358,7 +2358,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "AVRO struct key group by primitive",
       "statements": [
@@ -2403,7 +2403,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "AVRO group by struct",
       "statements": [
@@ -2474,7 +2474,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "JSON group by array",
       "statements": [
@@ -2494,7 +2494,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "JSON group by struct",
       "statements": [
@@ -2514,7 +2514,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "JSON group by struct convert key format",
       "statements": [
@@ -2554,7 +2554,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "JSON group by struct convert to incompatible key format",
       "statements": [
@@ -2568,7 +2568,7 @@
     },
     {
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "name": "Struct key used in aggregate expression",
       "statements": [

diff --git a/ksqldb-functional-tests/src/test/resources/query-validation-tests/json.json b/ksqldb-functional-tests/src/test/resources/query-validation-tests/json.json
@@ -279,7 +279,7 @@
     {
       "name": "ARRAY - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K ARRAY<DOUBLE> KEY, V INT) WITH (kafka_topic='input_topic', format='JSON');",
@@ -327,7 +327,7 @@
     {
       "name": "MAP - key - C*",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K MAP<INT, DOUBLE> KEY, foo INT) WITH (kafka_topic='input_topic', format='JSON');"
@@ -340,7 +340,7 @@
     {
       "name": "nested MAP type - key - C*",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K ARRAY<MAP<INT, DOUBLE>> KEY, foo INT) WITH (kafka_topic='input_topic', format='JSON');"
@@ -353,7 +353,7 @@
     {
       "name": "MAP - key - C*AS",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (k STRING KEY, v STRING) WITH (kafka_topic='input_topic', format='JSON');",
@@ -367,7 +367,7 @@
     {
       "name": "nested MAP type - key - C*AS",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (k STRING KEY, v STRING) WITH (kafka_topic='input_topic', format='JSON');",
@@ -411,7 +411,7 @@
     {
       "name": "STRUCT - key",
       "properties": {
-        "ksql.key.format.enabled": true
+        "ksql.complex.key.format.enabled": true
       },
       "statements": [
         "CREATE STREAM INPUT (K STRUCT<F DOUBLE> KEY, V INT) WITH (kafka_topic='input_topic', format='JSON');",