/datamarts/{datamart}/entities/{entity}/upload

Содержание раздела

Признак обновления/удаления записей (sys_op)
1. Как задать признак при потоковой загрузке
2. Различия между sysOp и sys_op
Как работает запрос
URL
Заголовки запроса
1. Заголовок x-request-id
2. Заголовок Content-Type
Тело запроса
Примеры запросов
Формат ответа

POST-метод выполняет потоковую загрузку данных в указанную таблицу.

Загрузка данных возможна в форматах CSV и Avro. Формат данных задается в заголовке Content-Type, а сами данные передаются в теле запроса.

Признак обновления/удаления записей (sys_op)

Системный признак sys_op позволяет определить порядок обработки записей при загрузке данных в логические таблицы и снапшот-таблицы.

Возможные значения sys_op:

0 — добавить новую запись или обновить текущую запись таблицы загруженными данными;
1 — удалить запись таблицы, соответствующую загруженной.

Как задать признак при потоковой загрузке

Чтобы задать значение признака sys_op при потоковой загрузке данных, используйте любой из способов:

[для всех записей] укажите опцию sysOp в URL-строке запроса;
[для каждой записи отдельно] в загружаемых данных определите для каждой записи поле sys_op со значением 0 или 1.

Различия между sysOp и sys_op

Порядок применения опции sysOp и поля sys_op различается, как показано в таблице ниже.

Способ задания	Применяется к записям	Приоритет применения
Опция `sysOp` в URL	Все записи потока данных	Высший
Поле `sys_op` записи	Конкретная запись	Средний
Значение не задано (включая пустое поле `sys_op`)	Все записи потока данных	Низший

При загрузке данных в прокси- и standalone-таблицы опция sysOp игнорируется, значение sys_op также игнорируется (если в таблице нет такого столбца) или сохраняется как есть (если столбец есть).

Как работает запрос

Обработка при разрыве соединения

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

Обработка загружаемых записей

Обработка каждой загружаемой записи зависит от наличия/отсутствия в таблице записи с таким же первичным ключом (PK), а также значений опций set.on.conflict.do и set.delete.tracking.enable таблицы (для снапшот-таблиц), как показано ниже.

Таблица	PK новый или таблица без PK	PK совпадает и sys_op=0	PK совпадает и sys_op=1
Логическая таблица	Добавление новой записи	Обновление записи	Удаление записи
Снапшот-таблица	Добавление новой записи	[set.on.conflict.do=update] обновление записи; [set.on.conflict.do=error] ошибка; [set.on.conflict.do=nothing] пропуск записи	[set.delete.tracking.enable=false] удаление записи; [set.delete.tracking.enable=true и set.on.conflict.do=error] ошибка; [set.delete.tracking.enable=true и set.on.conflict.do=nothing] пропуск записи
Прокси-, standalone-таблица	Добавление новой записи	[ADP] обновление записи; [ADB, ADQM, ADG] ошибка или вставка записи с дубликатом PK (зависит от СУБД)

Особенности загрузки в партиции

Загрузка в партиции имеет особенности:

[через партиционированную таблицу] записи распределяются по партициям; записи вне диапазонов партиций игнорируются;
[напрямую в партиции] записи вне диапазонов партиции игнорируются.

Заполнение столбцов таблицы

При загрузке данных столбцы заполняются:

[указанные в запросе столбцы] значениями из загружаемых данных;
[пропущенные nullable-столбцы] значениями, используемыми по умолчанию в СУБД.

Пропуск в запросе обязательных столбцов (NOT NULL) таблицы не поддерживается: возвращается ошибка.

Выполнение при отключенном датасорсе

Отключенный датасорс пропускается при исполнении запроса. Загрузка данных считается успешной, если данные сохранены в необходимые датасорсы.

URL

{baseUrl}/api/v1/datamarts/{datamart}/entities/{entity}/upload[?[sysOp={sysOpValue}][&queryId={queryIdValue}][&commitOnDisconnect={booleanValue}]]

Параметры

baseUrl — адрес ноды Prostore, состоящий из IP-адреса или доменного имени и номера порта;
datamart — имя логической базы данных, которой принадлежит целевая таблица. Для загрузки данных в standalone-таблицу укажите логическую БД, которой принадлежит связанная внешняя writable-таблица;
entity — имя таблицы, куда загружаются данные. Возможные значения:
- имя логической таблицы любого вида,
- имя снапшот-таблицы,
- имя прокси-таблицы,
- имя внешней writable-таблицы, связанной с целевой standalone-таблицей;
sysOpValue — значение опции sysOp;
queryIdValue — значение опции queryId;
booleanValue — значение опции commitOnDisconnect.

Опция sysOp

Устанавливает значение sys_op для всех загружаемых записей.

Опция предназначена для загрузки данных в логические таблицы и снапшот-таблицы; для прокси- и standalone-таблиц опция игнорируется. Опция имеет приоритет над индивидуальными значениями sys_op записей.

Возможные значения:

0 — вставить записи в таблицу, то есть добавить новые записи и (или) обновить текущие записи таблицы загруженными данными;
1 — удалить из таблицы записи, соответствующие загруженным.

Удаление данных с помощью потоковой загрузки возможно только из логических таблиц и снапшот-таблиц. Подробнее см. в разделе Основные понятия > Загрузка данных.

Опция queryId

Устанавливает идентификатор запроса SQL+, по которому можно отследить обработку запроса с помощью логов.

Опция commitOnDisconnect

Управляет загруженными данными и завершением операции записи при разрыве соединения во время потоковой загрузки, как показано в таблице ниже.

Целевая таблица	commitOnDisconnect=false (по умолчанию)	commitOnDisconnect=true
Логическая, снапшот-таблица	Буферы очищаются. Данные, загруженные в датасорсы, отменяются. Операция записи завершается как неуспешная	Данные из буферов догружаются в датасорсы. Данные, загруженные в датасорсы, фиксируются как текущее состояние данных. Операция записи завершается как успешная
Прокси-, standalone-таблица	Буферы очищаются. Данные, загруженные в датасорсы, остаются в них	Данные из буферов догружаются в датасорсы. Данные, загруженные в датасорсы, остаются в них

Заголовки запроса

Заголовок x-request-id

Опциональный заголовок задает уникальный идентификатор HTTP-запроса. Если не указан, система генерирует UUID-значение и возвращает его в качестве идентификатора в ответе.

Заголовок Content-Type

Обязательный заголовок определяет формат данных, содержащихся в теле запроса. Возможные значения:

text/csv — CSV-формат;
application/avro — Avro-формат.

Тело запроса

Тело запроса содержит загружаемые данные в CSV- или Avro-формате в виде строки.

Формат данных должен соответствовать формату, заданному в заголовке Content-Type. Данные должны иметь полное и корректное представление, то есть включать:

[CSV-формат] CSV-заголовок и строки данных;
[Avro-формат] Avro-файл, состоящий из заголовка, содержащего схему данных, и блоков данных.

Разделители, кавычки, символы экранирования и способ указания null-значений в CSV-данных должны соответствовать настроенным в секции конфигурации csvParser.

Avro-данные могут быть упакованы без сжатия или со сжатием по любому из алгоритмов. Степень сжатия и алгоритм указывать в запросе не нужно.

Примеры запросов

Загрузка данных в CSV-формате без дополнительных опций

Запрос без дополнительных опций, показанный ниже, вставляет записи с sys_op=0 и удаляет записи с sys_op=1. При разрыве соединения результаты загрузки отменяются.

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload' \
  -H 'accept: application/json' \
  -H 'x-request-id: f65ae2b4-7ee0-4e8b-aab4-fa856d32c723' \
  -H 'Content-Type: text/csv' \
  -d 'id;varchar_col;sys_op
     1;aaa;0
     2;bbb;1
     3;ccc;0'

Загрузка данных в CSV-формате с дополнительными опциями

Запрос на вставку данных (sysOp=0) с указанным идентификатором запроса (queryId=12345) и сохранением результатов загрузки при разрыве соединения (commitOnDisconnect=true):

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload?sysOp=0&queryId=12345&commitOnDisconnect=true' \
  -H 'accept: application/json' \
  -H 'x-request-id: 8600581d-adde-44ea-b691-a992027f928c' \
  -H 'Content-Type: text/csv' \
  -d 'id;varchar_col
     1;aaa
     2;bbb
     3;ccc'

Запрос на удаление данных (sysOp=1) с указанным идентификатором запроса (queryId=43345) и отменой результатов загрузки при разрыве соединения (commitOnDisconnect=false):

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload?sysOp=1&queryId=43345&commitOnDisconnect=false' \
  -H 'accept: application/json' \
  -H 'x-request-id: 4b6c9d58-d0e0-4911-96d4-838bf95319a5' \
  -H 'Content-Type: text/csv' \
  -d 'id;varchar_col
     1;aaa
     2;bbb
     3;ccc'

Загрузка данных в Avro-формате без дополнительных опций

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

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload' \
  -H 'accept: application/json' \
  -H 'x-request-id: 5db80212-db9d-437a-8e64-4815795b77a5' \
  -H 'Content-Type: application/avro' \
  -d 'string<AVRO>'

Загрузка данных в Avro-формате с дополнительными опциями

Запрос на вставку данных (sysOp=0) с указанным идентификатором запроса (queryId=76765) и сохранением результатов загрузки при разрыве соединения (commitOnDisconnect=true):

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload?sysOp=0&queryId=76765&commitOnDisconnect=true' \
  -H 'accept: application/json' \
  -H 'x-request-id: f1f7a335-e008-4665-9535-e5471305d131' \
  -H 'Content-Type: application/avro' \
  -d 'string<AVRO>'

Запрос на удаление данных (sysOp=1) с указанным идентификатором запроса (queryId=265384) и отменой результатов загрузки при разрыве соединения (commitOnDisconnect=false):

curl -X 'POST' \
  'http://localhost:9090/api/v1/datamarts/marketing/entities/sales/upload?sysOp=1&queryId=265384&commitOnDisconnect=false' \
  -H 'accept: application/json' \
  -H 'x-request-id: 2204bc82-714d-4f97-9e3e-fc7909547666' \
  -H 'Content-Type: application/avro' \
  -d 'string<AVRO>'

Формат ответа

Ответ возвращается в JSON-формате.

Варианты ответа:

неуспешный ответ (промежуточный или финальный),
промежуточный успешный ответ,
финальный успешный ответ.

Промежуточные ответы возвращаются только при загрузке по HTTP/2. Финальные ответы возвращаются как при загрузке по HTTP/2, так и по HTTP/1.1.

Неуспешный ответ

Неуспешный ответ возвращается:

при ошибке загрузки данных;
при отмене операции записи со стороны внешней системы;
при разрыве соединения, если запрос выполнялся с отключенной опцией commitOnDisconnect.

Неуспешный ответ содержит описание ошибки в exceptionMessage и возвращается с кодом:

500 — если проблема возникла до первого промежуточного успешного ответа;
200 — если проблема возникла во время или после первого промежуточного успешного ответа.

Пример неуспешного ответа с кодом 500:

{
  "exceptionMessage": "CSV values size [4] not equal to header size [3]"
}

Промежуточный успешный ответ

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

Пример промежуточного успешного ответа:

{
  "rowsReceivedTotal": 103
}

Финальный успешный ответ

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

при успешной загрузке всех порций данных из потока загрузки;
при разрыве соединения, если запрос выполнялся с включенной опцией commitOnDisconnect.

Пример финального успешного ответа:

{
  "queryId": "112233",
  "result": [
    {
      "sysCn": 1741301135352731,
      "ts": "2025-03-06 22:46:54.354123",
      "rowsAffected": 1206
    }
  ],
  "metadata": [
    {
      "name": "sysCn",
      "type": "BIGINT",
      "size": null,
      "nullable": false
    },
    {
      "name": "ts",
      "type": "TIMESTAMP",
      "size": 6,
      "nullable": false
    },
    {
      "name": "rowsAffected",
      "type": "BIGINT",
      "size": null,
      "nullable": false
    }
  ],
  "rows": 1,
  "statistics": {
    "elapsedTotalMs": 6338,
    "elapsedDbMs": 459
  }
}