implemented reading in rowbinary format for json and dynamic

Author: chernser

client-v2/src/main/java/com/clickhouse/client/api/data_formats/internal/BinaryStreamReader.java

@@ -22,6 +22,7 @@
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -92,7 +93,7 @@ public <T> T readValue(ClickHouseColumn column, Class<?> typeHint) throws IOExce
             }
         }
 
-        ClickHouseDataType dataType = column.getDataType() == ClickHouseDataType.Dynamic ? readDynamicValueDataType() : column.getDataType();
+        ClickHouseDataType dataType = column.getDataType() == ClickHouseDataType.Dynamic ? readDynamicData() : column.getDataType();
         int estimatedLen = column.getEstimatedLength();
         int precision = column.getPrecision();
         int scale = column.getScale();
@@ -199,7 +200,7 @@ public <T> T readValue(ClickHouseColumn column, Class<?> typeHint) throws IOExce
                     if (jsonAsString) {
                         return (T) readString(input);
                     } else {
-                        throw new RuntimeException("Reading JSON from binary is not implemented yet");
+                        return (T) readJsonData(input);
                     }
 //                case Object: // deprecated https://clickhouse.com/docs/en/sql-reference/data-types/object-data-type
                 case Array:
@@ -986,7 +987,7 @@ public byte[] allocate(int size) {
         }
     }
 
-    private ClickHouseDataType readDynamicValueDataType() throws IOException {
+    private ClickHouseDataType readDynamicData() throws IOException {
         byte tag = readByte();
 
         ClickHouseDataType type;
@@ -1005,4 +1006,21 @@ private ClickHouseDataType readDynamicValueDataType() throws IOException {
 
         return type;
     }
+
+    private static final ClickHouseColumn JSON_PLACEHOLDER_COL = ClickHouseColumn.parse("v Dynamic").get(0);
+
+    private Map<String, Object> readJsonData(InputStream input) throws IOException {
+        int numOfPaths = readVarInt(input);
+        if (numOfPaths == 0) {
+            return Collections.emptyMap();
+        }
+
+        Map<String, Object> obj = new HashMap<>();
+        for (int i = 0; i < numOfPaths; i++) {
+            String path = readString(input);
+            Object value = readValue(JSON_PLACEHOLDER_COL);
+            obj.put(path, value);
+        }
+        return obj;
+    }
 }

client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java

@@ -80,9 +80,11 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.regex.Matcher;
 import java.util.stream.BaseStream;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -1420,7 +1422,10 @@ private List<Map<String, Object>> prepareDataSet(String table, List<String> colu
             }
             createStmtBuilder.setLength(createStmtBuilder.length() - 2);
             createStmtBuilder.append(") ENGINE = MergeTree ORDER BY tuple()");
-            client.execute(createStmtBuilder.toString()).get(10, TimeUnit.SECONDS);
+            client.execute(createStmtBuilder.toString(), (CommandSettings)
+                    new CommandSettings().serverSetting("enable_dynamic_type", "1")
+                            .serverSetting("allow_experimental_json_type", "1"))
+                    .get(10, TimeUnit.SECONDS);
 
             // Insert data
             StringBuilder insertStmtBuilder = new StringBuilder();
@@ -2058,17 +2063,59 @@ public void testGettingRowsBeforeLimit() throws Exception {
     @Test(groups = {"integration"})
     public void testGetDynamicValue() throws Exception  {
         String table = "test_get_dynamic_values";
-        client.execute("DROP TABLE IF EXISTS " + table);
-        client.execute("CREATE TABLE " + table + " (rowId Int32, v Dynamic) Engine MergeTree ORDER BY ()", (CommandSettings) new CommandSettings().serverSetting("enable_dynamic_type", "1"));
 
-        client.execute("INSERT INTO " + table + " VALUES (0, 'string'), (1, 2222222)");
+        final AtomicInteger rowId = new AtomicInteger(-1);
+        final Random rnd = new Random();
+
+        List<Map<String,Object>> dataset = prepareDataSet(table, Arrays.asList("rowId Int32", "v Dynamic"),
+                Arrays.asList(s -> rowId.incrementAndGet(), s-> {
+                    int decision = rnd.nextInt(3);
+                    if (decision == 0) {
+                        return RandomStringUtils.randomAlphanumeric(3, 10);
+                    } else if (decision == 1) {
+                        return rnd.nextInt();
+                    } else {
+                        return rnd.nextDouble();
+                    }
+                }), 1000);
 
         try (QueryResponse response = client.query("SELECT * FROM " + table).get()) {
             ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);
-            reader.next();
-            System.out.println(reader.getString("v"));
-            reader.next();
-            System.out.println(reader.getString("v"));
+            while (reader.next() != null) {
+                int rowIndex  = reader.getInteger("rowId");
+                Assert.assertEquals(reader.getString("v"), dataset.get(rowIndex).get("v").toString());
+            }
+        }
+    }
+
+    @Test(groups = {"integration"})
+    public void testGetJSON() throws Exception  {
+        String table = "test_get_json_values";
+
+        final AtomicInteger rowId = new AtomicInteger(-1);
+        final Random rnd = new Random();
+
+        List<Map<String,Object>> dataset = prepareDataSet(table, Arrays.asList("rowId Int32", "v1 JSON"),
+                Arrays.asList(s -> rowId.incrementAndGet(),
+                s-> {
+                    String a = "{'a': '" + RandomStringUtils.randomAlphabetic(20) + "', 'b': { 'c': 'test1', 'd': " + rnd
+                            .nextInt(1000) + "}}";
+                    return a.replaceAll("'", "\"");
+                }), 1);
+
+        System.out.println(dataset);
+        ObjectMapper jackson = new ObjectMapper();
+        try (QueryResponse response = client.query("SELECT * FROM " + table).get()) {
+            ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response);
+            while (reader.next() != null) {
+                int rowIndex  = reader.getInteger("rowId");
+                JsonNode expected = jackson.readValue(dataset.get(rowIndex).get("v1").toString(), JsonNode.class);
+                Map<String, Object> v1 = reader.readValue("v1");
+                for (Map.Entry<String, Object> e : v1.entrySet()) {
+                    String pointer = "/" + e.getKey().replaceAll("\\.", "/");
+                    Assert.assertEquals(e.getValue().toString(), expected.at(pointer).asText());
+                }
+            }
         }
     }
 }