Support Protobuf insert #2187

public static void serializeProto(GeneratedMessageV3 message, ClickHouseColumn column, OutputStream out) throws IOException {

        Object value = null;

        // Get the descriptor of the field to serialize
        Descriptors.FieldDescriptor descriptor = getFieldDescriptor(message, column.getColumnName());

        if (descriptor != null) {
            if (descriptor.isRepeated() || message.hasField(descriptor)) {
                // Ensure the field has a value otherwise Protobuf returns the default value for type which is not necessarily
                // the default value ClickHouse uses in its table schema definition.
                value = message.getField(descriptor);

                LOGGER.debug("{}({}) - {} - nullable:{} - default:{}",
                        descriptor.getName(), descriptor.getType(), value, column.isNullable(), column.hasDefault());

                if (Set.of(ClickHouseDataType.DateTime, ClickHouseDataType.DateTime64).contains(column.getDataType())) {
                    // Some events DateTime fields are long. We need to convert them.
                    // We use `DateTime` and `DateTime64` only, but we may want to support more in the future.
                    // TODO: add more checking when needed as some Protobuf fields could be `com.google.protobuf.Timestamp`
                    value = convertFromInt((Long) value);
                }
            } else if (!message.hasField(descriptor) && !column.isNullable() && !column.hasDefault()) {
                // TODO: integrate this condition with the rest to also handle default dates
                // The field exists in the proto but it is not set. If it is not nullable or has no default value in
                // ClickHouse then we should take default proto value
                value = message.getField(descriptor);
            }
        } else {
            // In this case the protobuf does not have this column. In our case it's `insertTime` and nothing needs
            // to be done. This is an extra column not part of original data but set at run time by CH for further monitoring
            LOGGER.debug("{}({}) - {} - nulable:{} - default:{}",
                    column.getColumnName(), column.getDataType(), value, column.isNullable(), column.hasDefault());
        }

        // Same code as in `RowBinaryFormatWriter.commitRow`
        if (RowBinaryFormatSerializer.writeValuePreamble(out, true, column, value)) {
            ABSerializerUtils.serializeData(out, value, column);
        }
    }

else if (value instanceof GeneratedMessageV3) {
           GeneratedMessageV3 message = (GeneratedMessageV3) value;
           // Start: added section
           // From Protobuf messages Tuple are actually Protobuf messages themselves. So we need more capabilities to
           // translate them.
           for (ClickHouseColumn nestedColumn : column.getNestedColumns()) {
               // TODO: support cases when nested value is also a `DateTime` or `DateTime64` as above.
               Descriptors.FieldDescriptor descriptor = getFieldDescriptor(message, nestedColumn.getColumnName());
               Object nestedValue = message.getField(descriptor);
               serializeData(stream, nestedValue, nestedColumn);
           }
           // End: added section
       }

// Get ClickHouse table schema
TableSchema schema = clickhouseClient.getTableSchema(tableName);

// Serialise messages into an output stream
ByteArrayOutputStream out = new ByteArrayOutputStream();

for (GeneratedMessageV3 event: c.element().getValue()) {
    List<ClickHouseColumn> columnList = schema.getColumns();
    for (ClickHouseColumn column : columnList) {
        serializeProto(event, column, out);
    }
}

// Prepare insert
InputStream inputStream = new ByteArrayInputStream(out.toByteArray());
InsertSettings settings = new InsertSettings();
ClickHouseFormat format = ClickHouseFormat.RowBinaryWithDefaults;