Merge pull request #1794 from ClickHouse/clientv2_pojo_deserializers

[client-v2] POJO reader

Author: chernser

client-v2/pom.xml

@@ -85,6 +85,14 @@
             <scope>provided</scope>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/org.ow2.asm/asm -->
+        <dependency>
+            <groupId>org.ow2.asm</groupId>
+            <artifactId>asm</artifactId>
+            <version>9.7</version>
+        </dependency>
+
+
         <!-- Test dependencies -->
         <dependency>
             <groupId>${project.parent.groupId}</groupId>

client-v2/src/main/java/com/clickhouse/client/api/Client.java

@@ -20,8 +20,6 @@
 import com.clickhouse.client.api.insert.InsertResponse;
 import com.clickhouse.client.api.insert.InsertSettings;
 import com.clickhouse.client.api.insert.POJOSerializer;
-import com.clickhouse.client.api.insert.SerializerNotFoundException;
-import com.clickhouse.client.api.internal.BasicObjectsPool;
 import com.clickhouse.client.api.internal.ClickHouseLZ4OutputStream;
 import com.clickhouse.client.api.internal.ClientStatisticsHolder;
 import com.clickhouse.client.api.internal.ClientV1AdaptorHelper;
@@ -35,6 +33,7 @@
 import com.clickhouse.client.api.metrics.ClientMetrics;
 import com.clickhouse.client.api.metrics.OperationMetrics;
 import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.POJOSetter;
 import com.clickhouse.client.api.query.QueryResponse;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.api.query.Records;
@@ -70,7 +69,6 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -126,6 +124,10 @@ public class Client implements AutoCloseable {
     private final List<ClickHouseNode> serverNodes = new ArrayList<>();
     private final Map<Class<?>, List<POJOSerializer>> serializers; //Order is important to preserve for RowBinary
     private final Map<Class<?>, Map<String, Method>> getterMethods;
+
+    private final Map<Class<?>, Map<String, POJOSetter>> deserializers;
+    private final Map<Class<?>, Map<String, Method>> setterMethods;
+
     private final Map<Class<?>, Boolean> hasDefaults; // Whether the POJO has defaults
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
     private final ExecutorService sharedOperationExecutor;
@@ -143,9 +145,11 @@ private Client(Set<String> endpoints, Map<String,String> configuration, boolean
         this.endpoints.forEach(endpoint -> {
             this.serverNodes.add(ClickHouseNode.of(endpoint, this.configuration));
         });
-        this.serializers = new HashMap<>();
-        this.getterMethods = new HashMap<>();
-        this.hasDefaults = new HashMap<>();
+        this.serializers = new ConcurrentHashMap<>();
+        this.getterMethods = new ConcurrentHashMap<>();
+        this.hasDefaults = new ConcurrentHashMap<>();
+        this.deserializers = new ConcurrentHashMap<>();
+        this.setterMethods = new ConcurrentHashMap<>();
 
         boolean isAsyncEnabled = MapUtils.getFlag(this.configuration, ClickHouseClientOption.ASYNC.getKey());
         if (isAsyncEnabled && sharedOperationExecutor == null) {
@@ -929,61 +933,76 @@ public boolean ping(long timeout) {
      * @param clazz - class of a POJO
      * @param schema - correlating table schema
      */
-    public void register(Class<?> clazz, TableSchema schema) {
+    public synchronized void register(Class<?> clazz, TableSchema schema) {
         LOG.debug("Registering POJO: {}", clazz.getName());
 
         //Create a new POJOSerializer with static .serialize(object, columns) methods
-        List<POJOSerializer> serializers = new ArrayList<>();
-        Map<String, Method> getterMethods = new HashMap<>();
-
-        for (Method method: clazz.getMethods()) {//Clean up the method names
+        Map<String, Method> classGetters = new HashMap<>();
+        Map<String, Method> classSetters = new HashMap<>();
+        for (Method method : clazz.getMethods()) {//Clean up the method names
             String methodName = method.getName();
             if (methodName.startsWith("get") || methodName.startsWith("has")) {
                 methodName = methodName.substring(3).toLowerCase();
-                getterMethods.put(methodName, method);
-            } if (methodName.startsWith("is")) {
+                classGetters.put(methodName, method);
+            } else if (methodName.startsWith("is")) {
                 methodName = methodName.substring(2).toLowerCase();
-                getterMethods.put(methodName, method);
+                classGetters.put(methodName, method);
+            } else if (methodName.startsWith("set")) {
+                methodName = methodName.substring(3).toLowerCase();
+                classSetters.put(methodName, method);
             }
         }
-        this.getterMethods.put(clazz, getterMethods);//Store the getter methods for later use
+        this.getterMethods.put(clazz, classGetters);//Store the getter methods for later use
+        this.setterMethods.put(clazz, classSetters);//Store the setter methods for later use
+        this.hasDefaults.put(clazz, schema.hasDefaults());
 
+        List<POJOSerializer> classSerializers = new ArrayList<>();
+        Map<String, POJOSetter> classDeserializers = new ConcurrentHashMap<>();
         for (ClickHouseColumn column : schema.getColumns()) {
-            String columnName = column.getColumnName().toLowerCase().replace("_", "").replace(".","");
-            serializers.add((obj, stream) -> {
-                if (!getterMethods.containsKey(columnName)) {
-                    LOG.warn("No getter method found for column: {}", columnName);
-                    return;
-                }
-                Method getterMethod = this.getterMethods.get(clazz).get(columnName);
-                Object value = getterMethod.invoke(obj);
-                boolean hasDefaults = this.hasDefaults.get(clazz);
-
-                //Handle null values
-                if (value == null) {
-                    if (hasDefaults && !column.hasDefault()) {//Send this only if there is no default
-                        BinaryStreamUtils.writeNonNull(stream);
+            String propertyName = column.getColumnName().toLowerCase().replace("_", "").replace(".", "");
+
+            Method getterMethod = classGetters.get(propertyName);
+            boolean classHashDefaults = this.hasDefaults.get(clazz);
+            if (getterMethod != null) {
+                classSerializers.add((obj, stream) -> {
+                    Object value = getterMethod.invoke(obj);
+
+                    //Handle null values
+                    if (value == null) {
+                        if (classHashDefaults && !column.hasDefault()) {//Send this only if there is no default
+                            BinaryStreamUtils.writeNonNull(stream);
+                        }
+                        BinaryStreamUtils.writeNull(stream);//We send this regardless of default or nullable
+                        return;
                     }
-                    BinaryStreamUtils.writeNull(stream);//We send this regardless of default or nullable
-                    return;
-                }
 
-                //Handle default
-                if (hasDefaults) {
-                    BinaryStreamUtils.writeNonNull(stream);//Write 0
-                }
+                    //Handle default
+                    if (classHashDefaults) {
+                        BinaryStreamUtils.writeNonNull(stream);//Write 0
+                    }
 
-                //Handle nullable
-                if (column.isNullable()) {
-                    BinaryStreamUtils.writeNonNull(stream);//Write 0
-                }
+                    //Handle nullable
+                    if (column.isNullable()) {
+                        BinaryStreamUtils.writeNonNull(stream);//Write 0
+                    }
 
-                //Handle the different types
-                SerializerUtils.serializeData(stream, value, column);
-            });
+                    //Handle the different types
+                    SerializerUtils.serializeData(stream, value, column);
+                });
+            } else {
+                LOG.warn("No getter method found for column: {}", propertyName);
+            }
+
+            Method setterMethod = classSetters.get(propertyName);
+            String columnName = column.getColumnName();
+            if (setterMethod != null) {
+                classDeserializers.put(columnName, SerializerUtils.compilePOJOSetter(setterMethod, column));
+            } else {
+                LOG.warn("No setter method found for column: {}", propertyName);
+            }
         }
-        this.serializers.put(clazz, serializers);
-        this.hasDefaults.put(clazz, schema.hasDefaults());
+        this.serializers.put(clazz, classSerializers);
+        this.deserializers.put(clazz, classDeserializers);
     }
 
     /**
@@ -1023,21 +1042,30 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data)
      * @param tableName - destination table name
      * @param data  - data stream to insert
      * @param settings - insert operation settings
+     * @throws IllegalArgumentException when data is empty or not registered
      * @return {@code CompletableFuture<InsertResponse>} - a promise to insert response
      */
     public CompletableFuture<InsertResponse> insert(String tableName, List<?> data, InsertSettings settings) {
 
+        if (data == null || data.isEmpty()) {
+            throw new IllegalArgumentException("Data cannot be empty");
+        }
+
+        //Lookup the Serializer for the POJO
+        List<POJOSerializer> serializers = this.serializers.get(data.get(0).getClass());
+        if (serializers == null || serializers.isEmpty()) {
+            throw new IllegalArgumentException("No serializers found for class '" + data.get(0).getClass() + "'. Did you forget to register it?");
+        } else {
+            LOG.info("serializers: {}", serializers.size());
+        }
+
         String operationId = startOperation();
         settings.setOperationId(operationId);
         if (useNewImplementation) {
             globalClientStats.get(operationId).start(ClientMetrics.OP_DURATION);
         }
         globalClientStats.get(operationId).start(ClientMetrics.OP_SERIALIZATION);
 
-        if (data == null || data.isEmpty()) {
-            throw new IllegalArgumentException("Data cannot be empty");
-        }
-
         //Add format to the settings
         if (settings == null) {
             settings = new InsertSettings();
@@ -1046,12 +1074,6 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
         boolean hasDefaults = this.hasDefaults.get(data.get(0).getClass());
         ClickHouseFormat format = hasDefaults? ClickHouseFormat.RowBinaryWithDefaults : ClickHouseFormat.RowBinary;
 
-        //Lookup the Serializer for the POJO
-        List<POJOSerializer> serializers = this.serializers.get(data.get(0).getClass());
-        if (serializers == null || serializers.isEmpty()) {
-            throw new SerializerNotFoundException(data.get(0).getClass());
-        }
-
         if (useNewImplementation) {
             String retry = configuration.get(ClickHouseClientOption.RETRY.getKey());
             final int maxRetries = retry == null ? (int) ClickHouseClientOption.RETRY.getDefaultValue() : Integer.parseInt(retry);
@@ -1508,6 +1530,61 @@ public List<GenericRecord> queryAll(String sqlQuery) {
         }
     }
 
+    public <T> List<T> queryAll(String sqlQuery, Class<T> clazz) {
+        return queryAll(sqlQuery, clazz, null);
+    }
+
+    /**
+     * WARNING: Experimental API
+     *
+     * <p>Queries data and returns collection with whole result. Data is read directly to a DTO
+     *  to save memory on intermediate structures. DTO will be instantiated with default constructor or
+     *  by using allocator</p>
+     * <p>{@code class} should be registered before calling this method using {@link #register(Class, TableSchema)}</p>
+     * <p>Internally deserializer is compiled at the register stage. Compilation is done using ASM library by
+     *  writing a bytecode</p>
+     *
+     * @param sqlQuery - query to execute
+     * @param clazz - class of the DTO
+     * @param allocator - optional supplier to create new instances of the DTO.
+     * @throws IllegalArgumentException when class is not registered or no setters found
+     * @return List of POJOs filled with data
+     * @param <T>
+     */
+    public <T> List<T> queryAll(String sqlQuery, Class<T> clazz, Supplier<T> allocator) {
+        Map<String, POJOSetter> classDeserializers = deserializers.get(clazz);
+
+        if (classDeserializers == null || classDeserializers.isEmpty()) {
+            throw new IllegalArgumentException("No deserializers found for class '" + clazz + "'. Did you forget to register it?");
+        }
+
+        try {
+            int operationTimeout = getOperationTimeout();
+            QuerySettings settings = new QuerySettings().setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes);
+            try (QueryResponse response = operationTimeout == 0 ? query(sqlQuery, settings).get() :
+                    query(sqlQuery, settings).get(operationTimeout, TimeUnit.MILLISECONDS)) {
+                List<T> records = new ArrayList<>();
+                RowBinaryWithNamesAndTypesFormatReader reader =
+                        (RowBinaryWithNamesAndTypesFormatReader) newBinaryFormatReader(response);
+
+                while (true) {
+                    Object record = allocator == null ? clazz.getDeclaredConstructor().newInstance() : allocator.get();
+                    if (reader.readToPOJO(classDeserializers, record)) {
+                        records.add((T) record);
+                    } else {
+                        break;
+                    }
+                }
+
+                return records;
+            }
+        } catch (ExecutionException e) {
+            throw new ClientException("Failed to get query response", e.getCause());
+        } catch (Exception e) {
+            throw new ClientException("Failed to get query response", e);
+        }
+    }
+
     /**
      * <p>Fetches schema of a table and returns complete information about each column.
      * Information includes column name, type, default value, etc.</p>
@@ -1548,6 +1625,23 @@ public TableSchema getTableSchema(String table, String database) {
         }
     }
 
+    public TableSchema getTableSchemaFromQuery(String sql, String name) {
+        final String describeQuery = "DESC (" + sql + ") FORMAT " + ClickHouseFormat.TSKV.name();
+
+        int operationTimeout = getOperationTimeout();
+
+        try (QueryResponse response = operationTimeout == 0 ? query(describeQuery).get() :
+                query(describeQuery).get(getOperationTimeout(), TimeUnit.SECONDS)) {
+            return new TableSchemaParser().readTSKV(response.getInputStream(), name, getDefaultDatabase());
+        } catch (TimeoutException e) {
+            throw new ClientException("Operation has likely timed out after " + getOperationTimeout() + " seconds.", e);
+        } catch (ExecutionException e) {
+            throw new ClientException("Failed to get table schema", e.getCause());
+        } catch (Exception e) {
+            throw new ClientException("Failed to get table schema", e);
+        }
+    }
+
     /**
      * <p>Executes a SQL command and doesn't care response. Useful for DDL statements, like `CREATE`, `DROP`, `ALTER`.
      * Method however returns execution errors from a server or summary in case of successful execution. </p>

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

@@ -2,14 +2,12 @@
 
 import com.clickhouse.client.api.ClientException;
 import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
-import com.clickhouse.client.api.internal.BasicObjectsPool;
-import com.clickhouse.client.api.internal.MapUtils;
 import com.clickhouse.client.api.metadata.TableSchema;
 import com.clickhouse.client.api.query.NullValueException;
+import com.clickhouse.client.api.query.POJOSetter;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.config.ClickHouseClientOption;
 import com.clickhouse.data.ClickHouseColumn;
-import com.clickhouse.data.value.ClickHouseArrayValue;
 import com.clickhouse.data.value.ClickHouseGeoMultiPolygonValue;
 import com.clickhouse.data.value.ClickHouseGeoPointValue;
 import com.clickhouse.data.value.ClickHouseGeoPolygonValue;
@@ -31,18 +29,14 @@
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.NoSuchElementException;
 import java.util.TimeZone;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
-import java.util.function.Supplier;
 
 public abstract class AbstractBinaryFormatReader implements ClickHouseBinaryFormatReader {
 
@@ -79,6 +73,32 @@ protected AbstractBinaryFormatReader(InputStream inputStream, QuerySettings quer
 
     protected AtomicBoolean nextRecordEmpty = new AtomicBoolean(true);
 
+    public boolean readToPOJO(Map<String, POJOSetter> deserializers, Object obj ) throws IOException {
+        boolean firstColumn = true;
+
+        for (ClickHouseColumn column : columns) {
+            try {
+                Object val = binaryStreamReader.readValue(column);
+                if (val != null) {
+                    POJOSetter deserializer = deserializers.get(column.getColumnName());
+                    if (deserializer != null) {
+                        deserializer.setValue(obj, val);
+                    }
+                }
+                firstColumn = false;
+            } catch (EOFException e) {
+                if (firstColumn) {
+                    endReached();
+                    return false;
+                }
+                throw e;
+            } catch (Exception e) {
+                throw new ClientException("Failed to put value of '" + column.getColumnName() + "' into POJO", e);
+            }
+        }
+        return true;
+    }
+
     /**
      * It is still internal method and should be used with care.
      * Usually this method is called to read next record into internal object and affects hasNext() method.

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

@@ -85,14 +85,7 @@ private <T> T readValueImpl(ClickHouseColumn column) throws IOException {
                 // Primitives
                 case FixedString: {
                     byte[] bytes = readNBytes(input, column.getEstimatedLength());
-                    int end = 0;
-                    for (int i = 0; i < bytes.length; i++) {
-                        if (bytes[i] == 0) {
-                            end = i;
-                            break;
-                        }
-                    }
-                    return (T) new String(bytes, 0, end, StandardCharsets.UTF_8);
+                    return (T) new String(bytes, 0, column.getEstimatedLength(), StandardCharsets.UTF_8);
                 }
                 case String: {
                     int len = readVarInt(input);