Merge pull request #571 from climber73/add-simple-aggr-func-type-support

add SimpleAggregateFunction column type support

Author: chernser

CHANGELOG.md

@@ -1,3 +1,6 @@
+# 1.3.3
+* Added support of SimpleAggregateFunction column type
+
 # 1.3.2 
 Added support of writing JSON as string to a JSON column.
 Upgraded dependency `org.slf4j:slf4j-reload4j` from `2.0.13` to `2.0.17`

VERSION

@@ -1 +1 @@
-v1.3.2
+v1.3.3

src/main/java/com/clickhouse/kafka/connect/sink/db/mapping/Column.java

@@ -25,6 +25,10 @@
 @Builder
 @Getter
 public class Column {
+    private static final Pattern PARAMETRIZED_TYPE_PATTERN = Pattern.compile(".+\\(.+\\)");
+    private static final Pattern DECIMAL_TYPE_PATTERN = Pattern.compile("Decimal(?<size>\\d{2,3})?\\s*(\\((?<a1>\\d{1,}\\s*)?,*\\s*(?<a2>\\d{1,})?\\))?");
+    private static final Pattern SIMPLE_AGGREGATE_FUNCTION_TYPE_PATTERN = Pattern.compile("^SimpleAggregateFunction\\s*\\([^,]+,\\s*(.+)\\)$");
+
     private static final Logger LOGGER = LoggerFactory.getLogger(Column.class);
 
     private String name;
@@ -229,7 +233,7 @@ public static Column extractColumn(String name, String valueType, boolean isNull
 
                         ColumnBuilder variantTypeBuilder = Column.builder().type(typePrecisionAndScale.getT1());
 
-                        if (Pattern.compile(".+\\(.+\\)").asMatchPredicate().test(definition)) {
+                        if (PARAMETRIZED_TYPE_PATTERN.asMatchPredicate().test(definition)) {
                             variantTypeBuilder = variantTypeBuilder
                                     .precision(typePrecisionAndScale.getT2())
                                     .scale(typePrecisionAndScale.getT3());
@@ -251,6 +255,13 @@ public static Column extractColumn(String name, String valueType, boolean isNull
             return extractColumn(name, valueType.substring("LowCardinality".length() + 1, valueType.length() - 1), isNull, hasDefaultValue, isSubColumn);
         } else if (valueType.startsWith("Nullable")) {
             return extractColumn(name, valueType.substring("Nullable".length() + 1, valueType.length() - 1), true, hasDefaultValue, isSubColumn);
+        } else if (valueType.startsWith("SimpleAggregateFunction")) {
+            Matcher m = SIMPLE_AGGREGATE_FUNCTION_TYPE_PATTERN.matcher(valueType);
+            if (!m.matches()) {
+                throw new RuntimeException("can't parse SimpleAggregateFunction type");
+            }
+            String argType = m.group(1).trim();
+            return extractColumn(name, argType, isNull, hasDefaultValue, isSubColumn);
         }
 
         // We're dealing with a primitive type here
@@ -277,9 +288,7 @@ private static Tuple3<Type, Integer, Integer> dispatchPrimitiveWithPrecisionAndS
             precision = Integer.parseInt(scaleAndTimezone[0].trim());
             LOGGER.trace("Parsed precision of DateTime64 is {}", precision);
         } else if (type == Type.Decimal) {
-            final Pattern patter = Pattern.compile("Decimal(?<size>\\d{2,3})?\\s*(\\((?<a1>\\d{1,}\\s*)?,*\\s*(?<a2>\\d{1,})?\\))?");
-            Matcher match = patter.matcher(valueType);
-
+            Matcher match = DECIMAL_TYPE_PATTERN.matcher(valueType);
             if (!match.matches()) {
                 throw new RuntimeException("type doesn't match");
             }

src/test/java/com/clickhouse/kafka/connect/sink/db/mapping/ColumnTest.java

@@ -28,6 +28,18 @@ public void extractLowCardinalityNullableColumn() {
         assertEquals(Type.STRING, col.getType());
     }
 
+    @Test
+    public void extractSimpleAggregateFunctionColumn() {
+        Column col = Column.extractColumn(newDescriptor("SimpleAggregateFunction(sum, Int64)"));
+        assertEquals(Type.INT64, col.getType());
+    }
+
+    @Test
+    public void extractSimpleAggregateFunctionNullableColumn() {
+        Column col = Column.extractColumn(newDescriptor("SimpleAggregateFunction(sum, Nullable(Int64))"));
+        assertEquals(Type.INT64, col.getType());
+    }
+
     @Test
     public void extractArrayOfLowCardinalityNullableColumn() {
         Column col = Column.extractColumn(newDescriptor("Array(LowCardinality(Nullable(String)))"));