CREATE STREAM USERS (NAME STRING) WITH (...);

CREATE STREAM FOO AS SELECT ID FROM BAR;

-- Default config: ksql.persistence.wrap.single.values=false

-- creates a stream, picking up the system default of not wrapping
-- the serialized value is expected to not be wrapped. 
-- if the serialized value is wrapped it will likely result in a deserialization error.
CREATE STREAM IMPLICIT_SOURCE (ID INT) WITH (...);

-- override 'ksql.persistence.wrap.single.values' to true
-- the serialized value is expected to not be
``` ### Создание потоков и таблиц с помощью CREATE STREAM AS и CREATE TABLE AS

Операторы C*AS по умолчанию фиксируют значения `ksql.persistence.wrap.single.keys` и `ksql.persistence.wrap.single.values`.

Пользователи могут переопределить эти настройки, указав любое из свойств предложения `WITH` семейства `WRAP_SINGLE_XXX`.

Эти параметры будут сохранены как часть метаданных созданного источника.

Данные настройки контролируют сериализацию данных в источнике и их десериализацию последующими запросами.

Если указать любой из этих параметров, то при наличии многопольного связанного формата возникнет ошибка.

Например:

```sql
-- Конфигурация по умолчанию: ksql.persistence.wrap.single.values=false

-- Создаёт поток, используя системную настройку не обёртывать
-- Сериализованные значения в базовой теме не будут обёрнуты. 
CREATE STREAM IMPLICIT_SOURCE AS SELECT ID FROM S;

-- Переопределяет 'ksql.persistence.wrap.single.values' на true
-- Сериализованные значения будут обёрнуты.
CREATE STREAM EXPLICIT_SOURCE WITH(WRAP_SINGLE_VALUE=true) AS SELECT ID FROM S;

-- Возникает ошибка, так как схема значений многополевая
CREATE STREAM BAD_SOURCE WITH(WRAP_SINGLE_VALUE=true) AS SELECT ID, COST FROM S;

-- [В будущем, со структурированными ключами]
-- Явно устанавливает оба параметра на false.
-- Сериализованные ключи и значения не будут обёрнуты.
CREATE SOURCE SK_SOURCE WITH(WRAP_SINGLE_FIELDS=false) AS SELECT ID KEY, NAME STRING FROM S;

-- Будет использовать параметры сериализации SINK для определения, должны ли значения быть обёрнутыми или нет.
INSERT INTO SINK SELECT ID FROM SOURCE;

-- Также будет использовать параметры сериализации `SINK`, хотя в этом случае они будут обёрнуты, поскольку являются многопольными.
INSERT INTO SINK SELECT ID, NAME FROM SOURCE;

-- Будет использовать параметры сериализации SINK, чтобы определить, должны ли однополевые схемы быть обёрнутыми или нет.
INSERT INTO SINK (ID) VALUES (10);

-- Также будет использовать параметры сериализации `SINK`, хотя в этом случае они будут обёрнуты, поскольку являются многопольными.
INSERT INTO SINK VALUES (10, 'bob');

.. _ksql-persistence-wrap-single-keys:

---------------------------------
ksql.persistence.wrap.single.keys
---------------------------------

Устанавливает значение по умолчанию для свойства ``WRAP_SINGLE_KEY``, если оно не указано явно в операторах :ref:`CREATE TABLE <create-table>`, :ref:`CREATE STREAM <create-stream>`, :ref:`CREATE TABLE AS SELECT <create-table-as-select>` или :ref:`CREATE STREAM AS SELECT <create-stream-as-select>`.

Эту настройку можно переключить с помощью команды `SET`:

.. code:: sql

    SET 'ksql.persistence.wrap.single.keys'='true';

Для получения дополнительной информации см. разделы :ref:`CREATE TABLE <create-table>`, :ref:`CREATE STREAM <create-stream>`, :ref:`CREATE TABLE AS SELECT <create-table-as-select>` и :ref:`CREATE STREAM AS SELECT <create-stream-as-select>`.

.. note:: Эта настройка не влияет на форматы, которые не поддерживают какой-либо внешний объект или запись. Например, ``DELIMITED``.

.. _ksql-persistence-wrap-single-values:

-----------------------------------
ksql.persistence.wrap.single.values
-----------------------------------

Устанавливает значение по умолчанию для свойства ``WRAP_SINGLE_VALUE``, если оно не указано явно в операторах :ref:`CREATE TABLE <create-table>`, :ref:`CREATE STREAM <create-stream>`, :ref:`CREATE TABLE AS SELECT <create-table-as-select>` или :ref:`CREATE STREAM AS SELECT <create-stream-as-select>`.

Эту настройку можно переключить с помощью команды `SET`:

.. code:: sql

    SET 'ksql.persistence.wrap.single.values'='false';

Для получения дополнительной информации см. разделы :ref:`CREATE TABLE <create-table>`, :ref:`CREATE STREAM <create-stream>`, :ref:`CREATE TABLE AS SELECT <create-table-as-select>` и :ref:`CREATE STREAM AS SELECT <create-stream-as-select>`.

.. note:: Эта настройка не влияет на форматы, которые не поддерживают какой-либо внешний объект или запись. Например, ``DELIMITED``.

 +=========================+========================================================================================================+
 | WRAP_SINGLE_KEY         | Управляет тем, как ключи десериализуются, когда схема ключа содержит только одно поле.                |
 |                         |                                                                                                        |
 |                         | Настройка управляет тем, как KSQL будет десериализовать ключ записей в предоставленной ``KAFKA_TOPIC``.|
 |                         | Если установлено значение ``true``, KSQL ожидает, что поля (поля) будут сериализованы как именованные поля|
 |                         | внутри записи.                                                                                           |
 |                         | Если установлено значение ``false`` и ключ имеет схему с одним полем, KSQL ожидает, что поле будет       |
 |                         | сериализовано как анонимное значение.                                                                     |
 |                         | Установка значения ``false``, когда ключ является многополевой схемой, приведёт к ошибке               |
 |                         |                                                                                                        |
 |                         | Если не указано, используется системное значение по умолчанию, определённое :ref:`ksql-persistence-wrap-single-keys`|
 |                         | и по умолчанию равное ``false``.                                                                         |
 |                         |                                                                                                        |
 |                         | Примечание: установка этого свойства для форматов, которые не поддерживают перенос, например `DELIMITED`,|
 |                         | приведёт
``` **JSON**

Формат JSON поддерживает значения JSON.

Формат JSON поддерживает все типы данных KSQL (см. раздел «Типы данных»). Поскольку сам JSON не поддерживает тип map, KSQL сериализует MAP как объекты JSON. Из-за этого формат JSON может поддерживать только объекты MAP со строковыми ключами.

Сериализованный объект должен быть строкой, сериализованной в Kafka и содержащей допустимое значение JSON. Формат поддерживает объекты JSON и примитивы верхнего уровня, массивы и карты. Подробнее см. ниже.

**Объекты JSON**

Значения, которые являются объектами JSON, вероятно, наиболее распространены.

Например, учитывая оператор KSQL:

.. code:: sql

    CREATE STREAM x (ID BIGINT, NAME STRING, AGE INT) WITH (VALUE_FORMAT='JSON', ...);

И значение JSON:

.. code:: json

       {
         "id": 120,
         "name": "bob",
         "age": "49"
       }

KSQL десериализует поля объекта JSON в соответствующие поля потока.

*Примечание:* KSQL имеет специальную обработку схем с одним полем. KSQL поддерживает схемы с одним полем, которые сериализуются как именованное поле внутри объекта JSON или анонимное значение JSON. По умолчанию KSQL ожидает, что исходные данные будут иметь ключи с одним полем, которые были сериализованы как анонимные значения, а значения — как объекты JSON. Аналогично, по умолчанию KSQL сериализует выходные данные таким же образом. Обработкой однопольных схем KSQL можно управлять. Для получения дополнительной информации см. раздел single_field_wrapping.

**Примитивы верхнего уровня, массивы и карты**

Формат JSON поддерживает чтение и запись примитивов верхнего уровня, массивов и карт.

Например, если задан оператор KSQL только с одним полем в схеме значений и свойством WRAP_SINGLE_VALUE, установленным в false:

.. code:: sql

    CREATE STREAM x (ID BIGINT) WITH (VALUE_FORMAT='JSON', WRAP_SINGLE_VALUE=false, ...);

И значением JSON:

.. code:: json

       10

KSQL может десериализовать значения в поле ID потока.

При сериализации данных с одним полем KSQL может сериализовать поле как анонимное значение, если для WRAP_SINGLE_VALUE установлено значение false, например:

.. code:: sql

    CREATE STREAM y WITH (WRAP_SINGLE_VALUE=false) AS SELECT id FROM x;

Десериализация схем с одним ключевым полем также может контролироваться через свойство WRAP_SINGLE_KEY. Однако по умолчанию для схем с одним ключом используются анонимные значения.

Для получения дополнительной информации см. раздел ksql_single_field_wrapping.

----
**Avro**

Формат AVRO поддерживает двоичную сериализацию Avro всех типов данных KSQL, включая записи и примитивы верхнего уровня, массивы и карты.

Формату требуется, чтобы KSQL был настроен на хранение и извлечение схем Avro из |sr-long|. Для получения дополнительной информации см. раздел install_ksql-avro-schema.

**Записи Avro**

Записи Avro могут быть десериализованы в соответствующие схемы KSQL.

Например, задав оператор KSQL вида:

.. code:: sql

    CREATE STREAM x (ID BIGINT, NAME STRING, AGE INT) WITH (VALUE_FORMAT='JSON', ...);

И запись Avro, сериализованная со схемой:

.. code:: json

       {
         "type": "record",
         "namespace": "com.acme",
         "name": "UserDetails",
         "fields": [
           { "name": "id", "type": "long" },
           { "name": "name", "type": "string" }
           { "name": "age", "type": "int" }
         ]
       }

KSQL будет десериализовывать поля записи Avro в соответствующие поля потока. Обработка однополевых схем может контролироваться. Для получения дополнительной информации см. раздел «single_field_wrapping».

-------------------------------------
Основные примитивы, массивы и карты
-------------------------------------

Формат Avro поддерживает чтение и запись основных примитивов, массивов и карт.

Например, если дано выражение KSQL с единственным полем в схеме значений и свойством «WRAP_SINGLE_VALUE», установленным в «false»:

.. код:: sql

    CREATE STREAM x (ID BIGINT) WITH (VALUE_FORMAT='AVRO', WRAP_SINGLE_VALUE=false, ...);

И значение Avro, сериализованное со схемой:

.. код:: json

       {
         "type": "long"
       }

KSQL может десериализовать значения в поле «ID» потока.

При сериализации данных с одним полем KSQL может сериализовать поле как анонимное значение, если для «WRAP_SINGLE_VALUE» установлено значение «false», например:

.. код:: sql

    CREATE STREAM y WITH (WRAP_SINGLE_VALUE=false) AS SELECT id FROM x;

Сериализация однополевых ключевых схем также может управляться через свойство «WRAP_SINGLE_KEY». Однако следует отметить, что по умолчанию для однополевых ключевых схем используются анонимные значения.

Для получения дополнительной информации см. «ksql_single_field_wrapping».

=========================
Управление сериализацией
=========================

KSQL предлагает несколько механизмов управления сериализацией и десериализацией.

Основной механизм заключается в выборе формата сериализации при создании потока или таблицы и указании «VALUE_FORMAT» в предложении «WITH».

.. код:: sql

    CREATE TABLE x (F0 INT, F1 STRING) WITH (VALUE_FORMAT='JSON', ...);

Дополнительную информацию о форматах, поддерживаемых KSQL, см. в разделе «ksql_formats».

KSQL предоставляет некоторую дополнительную конфигурацию, позволяющую управлять сериализацией:

.. _single_field_wrapping
-------------------------
Однополевое (раз)обёртывание
-------------------------

.. note:: Формат «DELIMITED» не затрагивается однополевой распаковкой.


Управление десериализацией одиночных полей
==========================================

Когда KSQL десериализует запись Kafka в строку, ключ десериализуется в поля ключа, а значение записи десериализуется в поля значения.

По умолчанию KSQL ожидает, что любой ключ с однополевой схемой будет сериализован как анонимное значение, а не как именованное поле внутри записи, как это было бы в случае наличия нескольких полей.

И наоборот, по умолчанию KSQL ожидает, что любое значение с однополевой схемой было сериализовано как именованное поле в записи.

Значения по умолчанию для ключей и значений с одним полем различаются, поскольку они отражают наиболее распространённые варианты использования KSQL. Кроме того, любое изменение ключевой схемы, скорее всего, станет критическим изменением для любой существующей системы, так как оно изменит разделение данных. Поэтому эволюция ключевых схем обычно не вызывает беспокойства.

Если взять схему значений в качестве примера, значение с несколькими полями может выглядеть следующим образом в JSON:

.. код:: json

   {
      "id": 134,
      "name": "John"
   }
   
Если значение имело только поле «id», KSQL всё равно ожидал бы, что значение будет сериализовано в виде именованного поля, например:

.. код:: json

   {
      "id": 134
   }
  
Если ваши данные содержат только одно поле, и это поле не заключено в объект JSON (или запись Avro при использовании формата «AVRO»), вы можете использовать свойство «WRAP_SINGLE_VALUE» в предложении «WITH» ваших выражений :ref:`CREATE TABLE <create-table>` или :ref:`CREATE STREAM <create-stream>`. Установка свойства в «false» сообщит KSQL, что значение не упаковано, то есть пример выше будет числом JSON:

.. код:: json

     134
     
Например, следующее создаёт таблицу, в которой значения в базовой теме были сериализованы как анонимный номер JSON:

.. код:: sql

    CREATE TABLE x (ID INT) WITH (WRAP_SINGLE_VALUE=false, ...); **Контроль сериализации одиночных полей**

WRAP_SINGLE_FIELDS — это сокращение для установки обоих свойств.

Если в операторе явно не заданы ключ или значение, будут использоваться системные значения по умолчанию, определённые ksql.persistence.wrap.single.keys и ksql.persistence.wrap.single.values. Эти системные значения также можно изменить. Для получения дополнительной информации см. документацию ksql-persistence-wrap-single-keys и ksql-persistence-wrap-single-values.

**Сериализация отдельных полей при работе с KSQL**

Когда KSQL сериализует строку в запись Kafka, поле ключа сериализуется в ключ записи, а любое поле значения — в значение записи.

По умолчанию, если ключ имеет только одно поле, KSQL будет сериализовать это поле как анонимное значение, а не как именованное поле в записи, как это было бы, если бы полей было несколько.

И наоборот, если значение имеет только одно поле, KSQL сериализует это поле как именованное поле внутри записи.

Значения по умолчанию для однополевых ключей и значений различаются, поскольку они отражают наиболее распространённые варианты использования KSQL. Кроме того, любое изменение схемы ключа, скорее всего, приведёт к нарушению работы любой существующей системы, так как оно изменит разделение данных. Поэтому эволюция схем ключей обычно не вызывает беспокойства.

Рассмотрим в качестве примера следующие операторы:

.. code:: sql

    CREATE STREAM x (f0 INT, f1 STRING) WITH (VALUE_FORMAT='JSON', ...);
    CREATE STREAM y AS SELECT f0 FROM x;

Второй оператор определяет поток с единственным полем в значении, названным f0.

При записи результата в Kafka KSQL по умолчанию сохранит единственное поле как именованное поле объекта JSON (или записи Avro, если используется формат AVRO):

.. code:: json

   {
      "F0": 10
   }

Если требуется, чтобы значение было сериализовано как анонимное, например:

.. code:: json

   10

Можно использовать свойство WRAP_SINGLE_VALUE в своём операторе. Например:

.. code:: sql

    CREATE STREAM y WITH(WRAP_SINGLE_VALUE=false) AS SELECT f0 FROM x;

Аналогично можно использовать свойство WRAP_SINGLE_KEY для управления сериализацией ключей с одиночными полями или использовать WRAP_SINGLE_FIELDS в качестве сокращения для настройки обоих.

Если оператор явно не устанавливает перенос ключа или значения, используются системные значения по умолчанию, определённые ksql.persistence.wrap.single.keys и ksql.persistence.wrap.single.values. Их также можно изменить. Дополнительную информацию см. в документации ksql-persistence-wrap-single-keys и ksql-persistence-wrap-single-values. С (С РАСПОЛОЖЕНИЕМ ОДИНОЧНОГО КЛЮЧА = истина) КАК ВЫБРАТЬ ИДЕНТИФИКАТОР ИЗ S;

    -- приводит к ошибке, так как схема значения является многополевой
    СОЗДАТЬ ПОТОК BAD_SINK С (РАСПОЛОЖЕНИЕМ ОДИНОЧНОГО ЗНАЧЕНИЯ = истина) КАК ВЫБРАТЬ ИДЕНТИФИКАТОР, СТОИМОСТЬ ИЗ S;

* Подходящие детали будут добавлены в changelog.rst.*

# Последствия для совместимости

Тесты уже проведены, чтобы гарантировать, что это изменение не изменит схему или формат любых внутренних тем.

Устаревшие запросы, то есть те, которые были запущены в предыдущих версиях KSQL, имеют развёрнутые строковые ключи и оборачивают значения с однополевыми схемами.
Чтобы сохранить обратную совместимость, это должно продолжаться после завершения предложенной работы.

Обратная совместимость схемы значений поддерживается по умолчанию для ksql.persistence.wrap.single.values = истина. Это означает, что устаревшие запросы будут продолжать оборачивать однополевые схемы значений.

Обратная совместимость схем ключей поддерживается по умолчанию для ksql.persistence.wrap.single.keys = ложь. Это означает, что устаревшие запросы не будут оборачивать однополевые ключевые схемы.

## Последствия для производительности

Решение о десериализации и сериализации значений либо в обёрнутом, либо в развёрнутом виде принимается только при инициировании запроса. После этого классы serde работают с фиксированной схемой. Поэтому не должно быть никаких последствий для производительности, кроме стоимости сериализации и десериализации оболочки записи, если пользователь настроил запрос на их использование.

## Последствия для безопасности

Нет.

# Отклоненные альтернативы

## Умные десериализаторы

Вместо того чтобы контролировать десериализацию обёрнутых и развёрнутых однополевых схем с помощью семейства свойств WRAP_SINGLE_XXX, можно почти сделать так, чтобы десериализаторы могли определять во время выполнения и для каждой записи отдельно, содержит ли сериализованные данные анонимное значение или одно поле внутри записи.

Например, рассмотрим:

```sql
СОЗДАТЬ STREAM FOO (ИДЕНТИФИКАТОР INT) С (...);

{
   "ID": 10
}

10

-- Default config: ksql.persistence.deserialization.wrapped.single.value=false
-- Default config: ksql.persistence.serialization.wrap.single.value=false
 
-- создаёт поток:
--  WRAPPED_SINGLE_VALUES: указывает, что исходные данные имеют незаключённые в перенос значения
--  WRAP_SINGLE_VALUE: указывает, что последующие запросы будут заключать в перенос по умолчанию
CREATE STREAM UNWRAPPED_EXPLICIT_SOURCE (ID INT) WITH (WRAPPED_SINGLE_VALUES=false, WRAP_SINGLE_VALUE=true, ...);
  
-- создаёт поток:
--   ksql.persistence.deserialization.wrapped.single.value указывает, что исходные данные не заключены в перенос
--   ksql.persistence.serialization.wrap.single.values указывает, что последующие запросы не будут заключать в перенос значения по умолчанию
CREATE STREAM IMPLICIT_SOURCE (ID INT) WITH (...);
 
-- создаёт поток:
--   с несколькими полями:, поэтому KSQL знает, что они будут заключены в перенос.
--   WRAP_SINGLE_VALUE: указывает, что последующие запросы НЕ будут заключать в перенос значения по умолчанию
CREATE STREAM MULTI_FIELD_SOURCE (ID INT, NAME STRING) WITH (WRAP_SINGLE_VALUE=false, ...);
 
-- приведёт к ошибке, поскольку `WRAPPED_SINGLE_VALUES` не может быть `false` для схемы с несколькими полями.
CREATE STREAM BAD_SOURCE (ID INT, NAME STRING) WITH (WRAPPED_SINGLE_VALUES=false, ...);
 
-- KSQL знает, что исходные данные заключены в перенос, потому что источник помечен как таковой
-- сериализованное значение будет заключено в перенос из-за унаследованных свойств
-- последующие запросы будут заключать значения в перенос по умолчанию из-за унаследованных свойств
CREATE STREAM A AS SELECT ID FROM EXPLICIT_SOURCE;
 
-- KSQL знает, что исходные данные заключены в перенос, потому что источник помечен как таковой
-- сериализованные значения не будут заключены в перенос из-за предложения with
-- последующие запросы не будут заключать значения в перенос по умолчанию, из-за предложения with
CREATE STREAM B WITH(WRAP_SINGLE_VALUE=false) AS SELECT ID FROM EXPLICIT_SOURCE;
 
-- KSQL знает, что исходные данные не заключены в перенос, потому что источник помечен как таковой и имеет схему с одним полем
-- сериализованное значение не будет заключено в перенос из-за унаследованных свойств
-- последующие запросы не будут заключать значения в перенос по умолчанию из-за унаследованных свойств
CREATE STREAM C AS SELECT ID FROM IMPLICIT_SOURCE;
 
-- KSQL знает, что исходные данные не заключены в перенос, потому что источник помечен как таковой и имеет схему с одним полем
-- сериализованное значение будет заключено в перенос из-за предложения with
-- последующие запросы будут заключать значения в перенос по умолчанию из-за предложения with
CREATE STREAM D WITH(WRAP_SINGLE_VALUE=true) AS SELECT ID FROM IMPLICIT_SOURCE;
 
-- KSQL знает, что исходные данные заключены в перенос, так как у него несколько полей
-- сериализованное значение будет заключено в перенос, поскольку у него есть несколько полей
-- последующие запросы не будут заключать одиночные значения в перенос по умолчанию из-за унаследованных свойств
CREATE STREAM E AS SELECT ID, NAME FROM MULTI_FIELD_SOURCE;
 
-- KSQL знает, что исходные данные заключены в перенос, так как у него несколько полей
-- сериализованное значение будет заключено в перенос, поскольку у него есть несколько полей
-- последующие запросы будут заключать одиночные значения в перенос по умолчанию из-за предложения with
CREATE STREAM F WITH(WRAP_SINGLE_VALUE=true) AS SELECT ID,