Поддерживаемые типы данных#

Тракт Архивирование (ARCH) поддерживает определенные типы данных, которые могут быть использованы в реплицируемых DTO объектах.

Примитивные типы#

Примечание

Подробнее о ⁣⁣Logical Type avro – см. ⁣⁣спецификацию.

Особенности реализации TimeStamp#

Переключение форматов хранения даты осуществляется параметром timestampAsAvroLogicalType.

Параметр необязательный, значение по умолчанию false.

В этом случае форматирование работает, как и раньше: в формате "YYYY-MM-ddTHH:mm:ss.SSSSSS" в текущем часовом поясе, хранение в String.

["null", "string"]

В Postgres, Oracle значения микросекунд округляются до 4-х разрядов. В результате, происходят расхождения TimeStamp, которые фактически должны были быть одинаковыми.

В случае значения параметра true, значение даты и времени хранится в формате Avro long

Logical type: timestamp-micros

type:\[\"null\",{\"type\":\"long\",\"logicalType\":\"timestamp-micros\"}\]

Примечание:

Включение параметра в True приведет к форматированию времени в целевом виде. В этом случае нужно быть готовыми, к тому, чтобы удалить имеющиеся данные и провести повторную процедуру Init из-за расхождения в форматах.

Коллекции#

Общий подход к представлению операций над коллекциями#

Коллекции передаются с помощью структур данных, описывающих операцию над коллекцией.

Такая структура данных всегда содержит поля:

• op_type – тип операции. Может принимать следующие значения:

• replace_all – полностью заменить значения в коллекции;

• add – добавить в коллекцию новые элементы;

• remove – удалить элементы из коллекции.

• items_type – Опциональное значение, указывающее на тип данных для операций с однотипными ссылочными элементами, всегда пустое для примитивных коллекций, пустое для разнотипных ссылочных коллекций/map-структур, и заполненное типом для типизированных коллекций/map-структур;

• is_ordered – признак упорядоченной коллекции, применяется только для примитивных коллекций. Пустое для ссылочных коллекций/map-структур;

• items – удаляемые/заменяемые/добавляемые элементы. Тип зависит от типа структуры данных, описывающего коллекцию.

В рамках одной транзакции может происходить одновременно несколько операций над одной и той же коллекцией (например, удалить старых 2 элемента и вместо них добавить 4 новых), поэтому на схеме коллекции представляются именно массивом операций над коллекциями.

Пример исходного LDM с коллекцией:

<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.Address" primary-key="pk" version-property="version" abstract="false" embeddable="false">
    <property name="addressId" type="java.lang.Long" mandatory="true"/>
    <property name="employees" type="sbp.ts.pprbod.autotest.emu.mapper.entity.Employee" mandatory="false" collection="set" mapped-by="address"/>
    <property name="version" type="java.lang.Long" mandatory="false"/>
    <key name="pk">
        <property name="addressId"/>
    </key>
</class>

Поле employes в avro будет представлено на примере схемы представления коллекции:

{
"type": "record",
"name": "Address",
"namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity",
"doc": "{\"ReferenceProps\":[{\"key\":\"modelVersion\",\"value\":\"0.0.1\"}]}",
"fields": [
  {
    "name": "addressId",
    "type": [
      "null",
      "long"
    ]
  },
  {
    "name": "employees",
    "type": [
      "null",
      {
        "type": "array",
        "items": {
          "type": "record",
          "name": "CollectionOperation",
          "namespace": "com.sbt.pprbod.avro.classes",
          "fields": [
            {
              "name": "op_type",
              "type": "string"
            },
            {
              "name": "items",
              "type": {
                "type": "array",
                "items": {
                  "type": "record",
                  "name": "Reference",
                  "fields": [
                    {
                      "name": "key",
                      "type": "string"
                    },
                    {
                      "name": "type",
                      "type": [
                        "null",
                        "string"
                      ]
                    }
                  ]
                }
              }
            }
          ]
        }
      }
    ]
  },
  {
    "name": "version",
    "type": [
      "null",
      "long"
    ]
  }
]
}

Поддерживаемые типы коллекций#

Пример Коллекции ссылок с разнотипными значениями:

{"name":"items","type":
  {
    "type":"array","items":{
      "type":"record",
      "name":"Reference",
      "fields":[
        {"name":"key","type":"string"},
        {"name":"type","type":["null","string"]}
      ]
    }
  }
}

Пример Коллекции примитивов:

{"name":"items","type": {"type" : "map", "values" : [
  {"type":"string"},
  {"type":"bytes"},
  {"type":"int"},
  {"type":"long"},
  {"type":"float"},
  {"type":"double"},
  {"type":"boolean"}
  ]}}

Пример Map примитивных значений:

{"name":"items","type": {"type" : "map", "values" : [
  {"type":"string"},
  {"type":"bytes"},
  {"type":"int"},
  {"type":"long"},
  {"type":"float"},
  {"type":"double"},
  {"type":"boolean"}
  ]}}

Пример Map ссылочных значений с разнотипными значениями:

{"name": "items", "type": 
 {"type": "map", "values": 
   {
   "type": "record",
   "name": "Reference",
   "fields": [
      {"name": "key", "type": "string"},
      {"name": "type", "type": ["null", "string"]}
    ]
  }
 }
}

Особенности обработки коллекций#

В случае, когда ⁣ ⁣collectionField=[] (пустой) – ранее отправлялось как пустой массив операций над коллекциями, для ЦСП это означало, что ни одной операции над коллекцией не было совершено. Это не соответствует соглашению, поэтому коллекцию очистить было невозможно.

Изменено на:

collectionField=[
  {
    op_type=replace_all
    items=[]
  }
] – Заменить содержимое имеющейся коллекции на пустой массив.

Таким образом реализована очистка коллекций, поведение сервиса приведено в соответствие с соглашением.

Ссылки и ссылочные типы#

• Поддерживаются поля, хранящие ссылки на идентификаторы объектов с учетом допустимых ограничений на значение для полей идентификаторов.

• Поддерживаются вложенные (дочерние) объекты, такие объекты преобразуются в ссылки на объекты соответствующего типа. Подробнее – в разделе «Форматы обмена данными с ЦСП», подраздел «Формат ключей для объектов».

Cсылки представляются в тракте с помощью записи типа com.sbt.pprbod.avro.classes.Reference.

Пример исходной LDM со ссылкой:

<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.Employee" primary-key="pk" version-property="version" abstract="false" embeddable="false">
    <property name="address" type="sbp.ts.pprbod.autotest.emu.mapper.entity.Address" mandatory="false"/>
    ....
</class>

Схема с использованием ссылок:

{
"type": "record",
"name": "Employee",
"namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity",
"doc": "{\"ReferenceProps\":[{\"key\":\"modelVersion\",\"value\":\"0.0.1\"}]}",
"fields": [
  {
    "name": "address",
    "type": [
      "null",
      {
        "type": "record",
        "name": "Reference",
        "namespace": "com.sbt.pprbod.avro.classes",
        "fields": [
          {
            "name": "key",
            "type": "string"
          },
          {
            "name": "type",
            "type": [
              "null",
              "string"
            ]
          }
        ]
      }
    ]
  },
  ....
]
}

Embedded типы#

Что такое Embeddable типы на уровне JPA#

В JPA есть такая функциональность, как Embeddable типы. Эти типы позволяют вынести часть полей, которые физически должны храниться в одной таблице, в отдельную сущность.

Пример класса Employer:

@Entity
@Table
public class Employer {
    @Id
    private Long id;
    @Version
    private Long version;
    @Column
    private Date statusDate;
    @Embedded
    private EmployerInfo employerInfo;
    @Embedded
    private AgreementInfo agreementInfo;
    @Embedded
    private TariffInfo tariffInfo;
    @Column
    private Date payrollDate;
...

Embeddable класс EmployerInfo:

@Embeddable
public class EmployerInfo {
    @Column
    private String inn;
    @Column
    private String name;
    @Column
    private String phone;
    @Column
    private String email;
    @Column
    private String address;
...

Embeddable-класс AgreementInfo:

@Embeddable
public class AgreementInfo {
    @Column
    private String number;
    @Column
    private Date date;
...

Embeddable-класс TariffInfo:

@Embeddable
public class TariffInfo {
    @Column
    private Short paymentType;
    @Column
    private BigDecimal firstYearTariff;
    @Column
    private BigDecimal nextYearTariff;
...

В примере выше типы EmployerInfo, AgreementInfo, TariffInfo встраиваются в тип Employer и на прикладном уровне рассматриваются как отдельная сущность, однако на физическом уровне это будет одна таблица с полями:

Таким образом, на физическом уровне отдельных таблиц EmployerInf, AgreementInfo и TariffInfo не существует.

Embeddable LDM#

На LDM схеме Embeddable-типы выглядят как отдельные классы с пометкой embeddable="true".

Общие принципы формирования PprbDataContainer для объектов, содержащих embedded-поля#

Embeddable-сущности включаются в контейнер основного объекта в виде полей, представленных avro-контейнером, сгенерированным по схеме embedded-типа.

Для приведенной выше LDM для типа com.sbt.salary.emission.model.attaching.Employer будет сгенерирована следующая схема:

{
"type": "record",
"name": "Employer",
"namespace": "com.sbt.salary.emission.model.attaching",
"fields": [
  {
    "name": "statusDate",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "hashKey",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "payrollDate",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "id",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "tariffApplied",
    "type": [
      "null",
      "boolean"
    ]
  },
  {
    "name": "version",
    "type": [
      "null",
      "long"
    ]
  },
  {
    "name": "createDate",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "status",
    "type": [
      "null",
      "string"
    ]
  },
  {
    "name": "cardAttachApplication",
    "type": [
      "null",
      {
        "type": "record",
        "name": "Reference",
        "namespace": "com.sbt.pprbod.avro.classes",
        "fields": [
          {
            "name": "key",
            "type": "string"
          },
          {
            "name": "type",
            "type": [
              "null",
              "string"
            ]
          }
        ]
      }
    ]
  },
  {
    "name": "employerInfo",
    "type": [
      "null",
      {
        "type": "record",
        "name": "EmployerInfo",
        "namespace": "com.sbt.salary.emission.model.attaching",
        "fields": [
          {
            "name": "address",
            "type": ["null","string"]
          },
          {
            "name": "phone",
            "type": ["null","string"]
          },
          {
            "name": "name",
            "type": ["null","string"]
          },
          {
            "name": "inn",
            "type": ["null","string"]
          },
          {
            "name": "email",
            "type": ["null","string"]
          }
        ]
      }
    ]
  },
  {
    "name": "agreementInfo",
    "type": [
      "null",
      {
        "type": "record",
        "name": "AgreementInfo",
        "namespace": "com.sbt.salary.emission.model.attaching",
        "fields": [
          {
            "name": "date",
            "type": ["null","string"]
          },
          {
            "name": "number",
            "type": ["null","string"]
          }         
        ]
      }
    ]
  },
  {
    "name": "tariffInfo",
    "type": [
      "null",
      {
        "type": "record",
        "name": "TariffInfo",
        "namespace": "com.sbt.salary.emission.model.attaching",
        "fields": [
          {
            "name": "nextYearTariff",
            "type": ["null","long"]
          },
          {
            "name": "firstYearTariff",
            "type": ["null","long"]
          },
          {
            "name": "paymentType",
            "type": ["null","short"]
          }         
        ]
      }
    ]
  },
]
}

Обработка отсутствия и null-значений для ebmedded-полей в основном объекте#

Рассматриваемые события актуальны только для векторов тех классов, которые по LDM модели имеют embeddable-поля:

Принцип изменения отдельных полей внутри embedded-объекта#

Архивирование (ARCH) рассматривает весь embedded-объект, как атомарный и неизменяемый. То есть изменить в нем возможно что-либо только передав в соответствующем поле новый EmbeddedContainer. Контейнер содержит полное состояние всего embedded-поля с учетом изменений.

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

Поэтому в случае UpdateEvent (если обновление затронуло поля внутри встраиваемого объекта) или SnapshotEvent нужно передавать полное состояние всего встраиваемого объекта.

Включение коллекций Embedded в основной объект#

Коллекции embedded–типов напрямую не поддерживаются, однако их возможно реализовать используя сущность-расшивку, включающую в себя:

• Обратную ссылку на объект, к которому должен относится embedded;

• Непосредственно сам embedded;

• Генерируемый идентификатор.

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

Примечание – на уровне интеграции с ЦСП сущность-расшивка передается как обычный бизнес-объект. Архивирование (ARCH) и ЦСП не предусматривает трансформаций, позволяющих скрыть ее и привести данные к представлению уровнем выше.

Пример векторов с коллекцией embedded-сущностей:

На данный момент коллекции embedded-сущностей реализуются через сущность-разверстку com.sbt.salary.emission.model.attaching.EmployerClientElementReference, которая содержит ссылки на корневой объект и на embedded-сущность.

CreateEvent для коллекции embedded-сущности:

CreateEvent {
    alias = 'com.sbt.salary.emission.model.attaching.Employer',
    id = f0b338bd-07c2-4c70-8972-a94f72f70cbf,
    primitives = {
        statusDate = 2020-11-10 11:07:06.951,
        payrollDate = 2019-08-21 00:00:00.0,
        id = f0b338bd-07c2-4c70-8972-a94f72f70cbf0,
        version = 0,
        isTariffApplied = false,
        status = ACTUAL,
        employerInfo ={
                address = null,
                phone = null,
                name = null,
                inn = null,
                email = null     
        },
        agreementInfo ={
                date = null,
                number = null     
        },
        triffInfo ={
                nextYearTariff = null,
                firstYearTariff = null,
                paymentType = null   
      }
       
       
    },
    references = {
         cardAttachApplication = 1f248ec9-d900-41e3-b4dc-a5aa21845991,
    },
    primitiveCollections = {},
    referenceCollections = {
        clientrefcollection=[
            "1f248ec9-d900-41e3-b4dc-a5aa21845991",
            "1f248ec9-d900-41e3-b4dc-a5aa21845992"
        ]
    }
}
 
CreateEvent {
    alias = 'com.sbt.salary.emission.model.attaching.EmployerClientElementReference',
    id = 1f248ec9-d900-41e3-b4dc-a5aa21845991,
    primitives = {
        reference = {
            entityId = 123
        }
    },
    references = {
        backReference = "f0b338bd-07c2-4c70-8972-a94f72f70cbf0",
    },
    primitiveCollections = {},
    referenceCollections = {}
}
 
CreateEvent {
    alias = 'com.sbt.salary.emission.model.attaching.EmployerClientElementReference',
    id = 1f248ec9-d900-41e3-b4dc-a5aa21845992,
    primitives = {
        reference = {
            entityId = 456
        }
    },
    references = {
        backReference = "f0b338bd-07c2-4c70-8972-a94f72f70cbf0",
    },
    primitiveCollections = {},
    referenceCollections = {}
}

Пример итогового транспортного контейнера для коллекций:

PARTITION: 2
{
container_type=initggregate
init_container=INIT CONTAINERS:
INIT CONTAINER: {
    packet=DataChunk
    data_container_count=3
    data_chunk=2
    zone_id=UNKNOWN_ZONE
    data_type=PprbData_1.0
    message_id=357
    pprbod_client_id=27335173-c648-403e-8e95-13533d89597b
    message_timestamp=1636732690460
    event_date=2021-11-12 18:58:10.000458
    global_key=7028545540438622209
    global_type=com.sbt.salary.emission.model.attaching.Employer
    global_version=5
    data_container=DATA CONTAINERS:
    DATA CONTAINER: {
        key=7028545540438622209
        entry_type=com.sbt.salary.emission.model.attaching.Employer
        version=5
        oper_type=I
        upd_attrs=[]
        avro_entry=DATA ENTRY: {
            hashKey=null
            id=f0b338bd-07c2-4c70-8972-a94f72f70cbf
            payrollDate=2019-08-21 00:00:00.000000
            status=ACTUAL
            statusDate=2020-11-10 11:07:06.951000
            tariffApplied=null
            clientrefcollection=[
                {
                    type=BookStoreClientElementReference
                    key=1f248ec9-d900-41e3-b4dc-a5aa21845991
                },
                {
                    type=BookStoreClientElementReference
                    key=1f248ec9-d900-41e3-b4dc-a5aa21845992
                }
            ]
        }
    }
  
    DATA CONTAINER: {
        key=1f248ec9-d900-41e3-b4dc-a5aa21845992
        entry_type=com.sbt.salary.emission.model.attaching.EmployerClientElementReference
        version=5
        oper_type=I
        upd_attrs=[]
        avro_entry=DATA ENTRY: {
                backReference={
                    type=com.sbt.salary.emission.model.attaching.Employer
                    key=7028545540438622209
                }
                clientReference={
                      entityId=123
                }
        }
    } 

  
    DATA CONTAINER: {
        key=1f248ec9-d900-41e3-b4dc-a5aa21845991
        entry_type=BookStoreClientElementReference
        version=5
        oper_type=I
        upd_attrs=[]
        avro_entry=DATA ENTRY: {
                backReference={
                    type=com.sbt.salary.emission.model.attaching.Employer
                    key=7028545540438622209
                }
                clientReference={
                    entityId=456
                }
        }
    }
        
    cont_hash=java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
}
cont_hash=java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
}

Пример вектора Прикладного журнала (APLJ) для объекта, содержащего Embedded-поля и PprbDataContainer, построенный из него#

Для приведенной выше LDM пример передачи Employer будет выглядеть следующим образом:

Исходный вектор Прикладного журнала (APLJ):

{
"serializerInfo": {
  "id": "3375153340802439707",
  "name": "json_gson",
  "format": "JSON"
},
"data": {
  "txId": null,
  "partitionId": null,
  "changeSets": [
    {
      "createEvents": [
        {
          "alias": "com.sbt.salary.emission.model.attaching.Employer",
          "id": "1611397353559275169",
          "primitives": {
            "statusDate": "2021-11-21 12:24:42.859000",
            "hashKey": "com.sbt.salary.emission.model.attaching.Employer#1611397353559275169",
            "payrollDate": "2021-11-21 12:24:42.859000",
            "id": "1611397353559275169",
            "tariffApplied": true,
            "version": 1,
            "createDate": "2021-11-21 12:24:42.859000",
            "status": "SUCCESS",
            "employerInfo": {
              "address": "г. Москва, ул. Свободы, д. 123",
              "phone": "8 (800)-123-4567",
              "name": "ПАО \"СельхозСтройБанк\"",
              "inn": "1234567891012",
              //email null!
            },
            "agreementInfo": null, //null весь embeddable!

            //tarifInfo не передан совсем
          },
          "references": {
            "cardAttachApplication": "2651396353559275154"
          },
          "primitiveCollections": {
          },
          "referenceCollections": {
          },
          "version": 1
        }
      ],
      "updateEvents": [],
      "deleteEvents": [],
      "snapshotEvents": []
    }
  ]
}
}

Пример Авро-контейнер для ЦСП:

{
zone_id=
data_type=PprbData_1.0
message_id=2021-11-20T21:00:00.000Z_1417729817790156734__key_null
pprbod_client_id=STREAM_{"consumerResponse":{"result":null,"errorMessage":null,"pluginCode":"EXPORT_FD4","partitionDate":1637442000000,"exportStatusId":3739236527447,"journalId":3573497225636,"zonedId":"CONSENTS","serviceId":"526e12174e254b239021b21669bdbd15","serviceIdType":null,"dataType":"UCP"},"zoneId":"CONSENTS","dataType":"UCP","pluginCode":"EXPORT_FD4","key":"1417729817790156734","topic":"journal_request_topic_CONSENTS_UCP","partition":6,"offset":2934}
message_timestamp=1637486683897
event_date=2021-11-21 12:24:43.000553
global_key=1611397353559275169
global_type=com.sbt.salary.emission.model.attaching.Employer
global_version=1
data_container=DATA CONTAINERS:
DATA CONTAINER: {
    key=1611397353559275169
    entry_type=com.sbt.salary.emission.model.attaching.Employer
    version=1
    oper_type=null
    upd_attrs=null
    avro_entry=DATA ENTRY: {
        statusDate=2021-11-21 12:24:42.859000
        hashKey=com.sbt.salary.emission.model.attaching.Employer#1611397353559275169
        payrollDate=2021-11-21 12:24:42.859000
        id=1611397353559275169
        tariffApplied=true
        version=1
        createDate=2021-11-21 12:24:42.859000
        status=SUCCESS
        cardAttachApplication={
            key=2651396353559275154
            type=com.sbt.salary.emission.model.attaching.CardAttachApplication
        }
        employerInfo={
            address=г. Москва, ул. Свободы, д. 123
            phone=8 (800)-123-4567
            name=ПАО "СельхозСтройБанк"
            inn=1234567891012
            email=null // незаполненное поле внутри embedded-а -> null
        }
        agreementInfo=null //null -> null
        tariffInfo=null //отсутствие -> null
          
    }
}
cont_hash=java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
}

Пример вектора Прикладного журнала (APLJ) для объекта, содержащего Embedded-ключ и PprbDataContainer, построенный из него#

Для приведенной выше LDM пример передачи Employer будет выглядеть следующим образом:

Исходный вектор Прикладного журнала (APLJ):

{
"type": "DATASPACE",
"txId": "86f13993-48bf-40c6-b3c4-0e21950b2a11",
"headers": {
  "rootClass": "sbp.com.sbt.dataspace.jpa.Product",
  "rootId": "7001023693027540993",
  "rootVersion": 1,
  "txTimestamp": 1630643832229
},
"partitions": [
  {
    "type": "ORM_CV",
    "serializer": "change_vector_json",
    "format": "JSON",
    "payload": {
      "serializerInfo": {
        "id": "3375153340802439707",
        "name": "json_gson",
        "format": "JSON"
      },
      "data": {
        "txId": null,
        "partitionId": null,
        "changeSets": [
          {
            "createEvents": [
              {
                "alias":"com.sbt.ae.spark.entity.access.AccessRightEntity",
                "id":{
                  "rightName":"LOADER_EXCEL",
                  "roleCode":"ae-spark_AGR.SPARK.Worker"
                },
                  "primitives":{
                    "userIns":"Kъgпфx",
                    "userUpd":"Рahыдeщw5-АT",
                    "dateUpd":"2021-08-27T11:27:32.837+03:00",
                    "dateIns":"2021-08-27T11:27:32.837+03:00",
                    "nVer":0},
                  "references":{
                    "accessRole":"ae-spark_AGR.SPARK.Worker"
                  },
                  "primitiveCollections":{},
                  "referenceCollections":{}
                }
            ],
            "updateEvents": [],
            "deleteEvents": [],
            "snapshotEvents": []
          }
        ]
      }
    }
  }
]
}

Авро-контейнер для ЦСП:

{
packet=EOD
data_container_count=2
data_chunk=1
zone_id=
data_type=PprbData_1.0
message_id=1
pprbod_client_id=e3fb8cd6-4e9d-438f-b291-f2093ee18aba
message_timestamp=1642402765316
event_date=2022-01-17 09:59:25.000300
global_key=7001023693027540993
global_type=sbp.com.sbt.dataspace.jpa.Product
global_version=1
data_container=DATA CONTAINERS:
DATA CONTAINER: {
  key=LOADER\_EXCEL_ae-spark\_AGR\.SPARK\.Worker
  entry_type=com.sbt.ae.spark.entity.access.AccessRightEntity
  version=1
  oper_type=I
  upd_attrs=[]
  avro_entry=DATA ENTRY: {
    accessRole={
      key=ae-spark_AGR.SPARK.Worker
      type=com.sbt.ae.spark.entity.access.AccessRole
    }
    dateIns=2021-08-27 11:27:32.837000
    dateUpd=2021-08-27 11:27:32.837000
    idPK={
      rightName=LOADER_EXCEL
      roleCode=ae-spark_AGR.SPARK.Worker
    }
    nVer=0
    reserved=null
    rightName=null
    roleCode=null
    userIns=Kъgпфx
    userUpd=Рahыдeщw5-АT
  }
}
cont_hash=java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
}

Пример вектора Прикладного журнала (APLJ) для объекта, содержащего коллекцию ссылок на объекты с композитными Embedded-ключами и PprbDataContainer, построенного из него#

Для приведенной выше LDM пример передачи Employer будет выглядеть следующим образом:

Исходный вектор Прикладного журнала (APLJ):

{
"type": "DATASPACE",
"txId": "86f13993-48bf-40c6-b3c4-0e21950b2a11",
"headers": {
  "rootClass": "sbp.com.sbt.dataspace.jpa.Product",
  "rootId": "7001023693027540993",
  "rootVersion": 1,
  "txTimestamp": 1630643832229
},
"partitions": [
  {
    "type": "ORM_CV",
    "serializer": "change_vector_json",
    "format": "JSON",
    "payload": {
      "serializerInfo": {
        "id": "3375153340802439707",
        "name": "json_gson",
        "format": "JSON"
      },
      "data": {
        "txId": null,
        "partitionId": null,
        "changeSets": [
          {
            "createEvents": [
              {
                "alias":"com.sbt.ae.spark.entity.oper.AmOper",
                "id":16621,
                "primitives":{
                  "date":"2021-08-27T11:27:32.837+03:00",
                  "keyAttrValues":39222,
                  "userUpd":"Undefined",
                  "dateIns":"2021-08-27T11:27:32.837+03:00",
                  "nVer":1,
                  "packageId":1634,
                  "idDeal":66,
                  "idOperDict":10544,
                  "idFileLine":16968,
                  "userIns":"Undefined",
                  "lock":4,
                  "dateUpd":"2021-08-27T11:27:32.837+03:00",
                  "status":3},
                "references":{
                  "owner":1,
                  "deal":66,
                  "subProdDict":1174,
                  "fileLine":16968,
                  "operDict":10544,
                  "packageOfEvents":1634,
                  "mapSelect":121403},
                "primitiveCollections":{},
                "referenceCollections":{
                  "amOperAttrs":[
                    {"attrDictId":14597, "operId":16621},
                    {"attrDictId":15116, "operId":16621},
                    {"attrDictId":14666, "operId":16621},
                    {"attrDictId":15020, "operId":16621},
                    {"attrDictId":15251, "operId":16621},
                    {"attrDictId":15255, "operId":16621},
                    {"attrDictId":15088, "operId":16621},
                    {"attrDictId":14665, "operId":16621},
                    {"attrDictId":14615, "operId":16621},
                    {"attrDictId":15164, "operId":16621},
                    {"attrDictId":15135, "operId":16621},
                    {"attrDictId":14601, "operId":16621},
                    {"attrDictId":14687, "operId":16621},
                    {"attrDictId":15218, "operId":16621}]}}
            ],
            "updateEvents": [],
            "deleteEvents": [],
            "snapshotEvents": []
          }
        ]
      }
    }
  }
]
}

Авро-контейнер для ЦСП:

{
packet=EOD
data_container_count=2
data_chunk=1
zone_id=
data_type=PprbData_1.0
message_id=1
pprbod_client_id=3fb98ea8-40bc-4cb3-89a8-21893b1e6d8f
message_timestamp=1642403382331
event_date=2022-01-17 10:09:42.000331
global_key=7001023693027540993
global_type=sbp.com.sbt.dataspace.jpa.Product
global_version=1
data_container=DATA CONTAINERS:
DATA CONTAINER: {
  key=16621
  entry_type=com.sbt.ae.spark.entity.oper.AmOper
  version=1
  oper_type=I
  upd_attrs=[]
  avro_entry=DATA ENTRY: {
    amOperAttrs=[
      {
        op_type=replace_all
        items=[
          {
            key=14597_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15116_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=14666_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15020_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15251_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15255_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15088_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=14665_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=14615_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15164_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15135_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=14601_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=14687_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
          {
            key=15218_16621
            type=com.sbt.ae.spark.entity.oper.AmOperAttr
          }
        ]
      }
    ]
    date=2021-08-27 11:27:32.837000
    dateIns=2021-08-27 11:27:32.837000
    dateUpd=2021-08-27 11:27:32.837000
    deal={
      key=66
      type=com.sbt.ae.spark.entity.TamDealEntity
    }
    eventStatus=null
    fileLine={
      key=16968
      type=com.sbt.ae.spark.entity.LoaderFileLine
    }
    id=16621
    idDeal=66
    idFileLine=16968
    idOperDict=10544
    keyAttrValues=39222
    lock=4
    mapSelect={
      key=121403
      type=com.sbt.ae.spark.entity.mapping.MapSelect
    }
    nVer=1
    operDict={
      key=10544
      type=com.sbt.ae.spark.entity.TpbOperDict
    }
    owner={
      key=1
      type=com.sbt.ae.spark.entity.TpbOwner
    }
    packageId=1634
    packageOfEvents={
      key=1634
      type=com.sbt.ae.spark.entity.events.PackageOfEvents
    }
    parentOper=null
    reserved=null
    status=3
    subProdDict={
      key=1174
      type=com.sbt.ae.spark.entity.TpbSubProdDict
    }
    userIns=Undefined
    userUpd=Undefined
  }
}
cont_hash=java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
}

Embedded-типы в составе других объектов#

Embedded-типы в составе других объектов представлены в виде record-записей, содержащих поля встраиваемого в него embedded-типа.

Пример LDM Embedded:

<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.Employee" primary-key="pk" version-property="version" abstract="false" embeddable="false">
    <property name="employeeId" type="java.lang.Long" mandatory="true"/>
    <property name="passport" type="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.Passport" mandatory="false"/>
    <property name="phoneNumber" type="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.PhoneNumber" mandatory="false"/>
    <property name="version" type="java.lang.Long" mandatory="false"/>
    <property name="workNumber" type="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.WorkNumber" mandatory="false"/>
    <key name="pk">
        <property name="employeeId"/>
    </key>
</class>
<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.Passport" abstract="false" embeddable="true">
    <property name="passportNumber" type="java.lang.Integer" mandatory="false"/>
    <property name="series" type="java.lang.String" mandatory="false"/>
</class>
<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.PhoneNumber" abstract="false" embeddable="true">
    <property name="number" type="java.lang.Integer" mandatory="false"/>
</class>
<class name="sbp.ts.pprbod.autotest.emu.mapper.entity.embedd.WorkNumber" abstract="false" embeddable="true">
    <property name="workNumber" type="java.lang.Integer" mandatory="false"/>
</class>

Схема с embedded типами:

{
"type": "record",
"name": "Employee",
"namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity",
"doc": "{\"ReferenceProps\":[{\"key\":\"modelVersion\",\"value\":\"0.0.1\"}]}",
"fields": [
  {
    "name": "employeeId",
    "type": [
      "null",
      "long"
    ]
  },
  {
    "name": "passport",
    "type": [
      "null",
      {
        "type": "record",
        "name": "Passport",
        "namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity.embedd",
        "fields": [
          {
            "name": "passportNumber",
            "type": [
              "null",
              "int"
            ]
          },
          {
            "name": "series",
            "type": [
              "null",
              "string"
            ]
          }
        ]
      }
    ]
  },
  {
    "name": "phoneNumber",
    "type": [
      "null",
      {
        "type": "record",
        "name": "PhoneNumber",
        "namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity.embedd",
        "fields": [
          {
            "name": "number",
            "type": [
              "null",
              "int"
            ]
          }
        ]
      }
    ]
  },
  {
    "name": "version",
    "type": [
      "null",
      "long"
    ]
  },
  {
    "name": "workNumber",
    "type": [
      "null",
      {
        "type": "record",
        "name": "WorkNumber",
        "namespace": "sbp.ts.pprbod.autotest.emu.mapper.entity.embedd",
        "fields": [
          {
            "name": "workNumber",
            "type": [
              "null",
              "int"
            ]
          }
        ]
      }
    ]
  }
]
}

Embedded типы в качестве составных ключей#

В тракте возможно использование Embdedded-объектов в качестве ключей (например Источник в прикладном коде использует EmbeddedId-аннотацию). В таком случае в качестве ключа для такого объекта выступает строка, в которую входят строковые представления полей данного embedded-объекта, соединенные через символ _ (подчеркивание).

Порядок полей соответствует алфавитному порядку следования их имен.

Включение Embedded-сущностей в основной объект#

При включении embedded-сущностей в основной объект используется вложенный avro-запись, содержащий в своей схеме поля embedded-объекта (пример схемы можно найти на основной странице про embedded)

Использование Embedded-сущности в качестве ключа#

При использовании Embedded-сущности в качестве ключа, формирование ключа в строковом виде подчиняется тем же правилам, по которым аналогичное строчное представление формируется для композитного ключа. То есть в ключ включаются поля Embedded-сущности, разделенные символом _ в алфавитном порядке.

Агрегаты Platform V DataSpace (APT)#

Описание#

Агрегат в DataSpace представляет собой древовидный объект, позволяющий хранить бесконечно большие объемы данных за счет своей структуры. Сами объекты в DataSpace небольшого размера, однако их может быть много, и они связаны с помощью обратных ссылок (дочерний объект содержит ссылку на родителя, родитель при этом информации о своих дочерних объектах не содержит). Засчет того, что глубина такого дерева может быть достаточно большой – суммарный объем сериализованных данных о всех взаимосвязанных таким способом объектах в агрегате может достигать нескольких Гигабайт.

Пример:

Агрегат организация "СберТех", в котором содержится информация о его финансовых действиях, о сотрудниках, об имуществе.

Сам объект СберТех может содержать только поля ИНН, название, год основания, и тд., и занимать незначительный объем. Однако на него ссылаются дочерние объекты типа "Отдел", на каждый объект типа "Отдел" ссылаются объекты типа "Команда", на Команды в свою очередь ссылаются сотрудники, на них – закрепленное имущество, данные о зарплате и тд.

Каждый объект небольшой по размеру, однако в сумме они дают внушительный объем:

Особенности передачи#

При передаче агрегата, если добавляется новая дочерняя сущность или меняется существующая – меняется версия всего агрегата.

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

Пример передачи версии агрегата:

{
"type": "DATASPACE",
"txId": "86f13993-48bf-40c6-b3c4-0e21950b2a11",
"headers": {
  "rootClass": "sbp.com.sbt.dataspace.jpa.Product",
  "rootId": "7001023693027540993",
  "rootVersion": 1,
  "txTimestamp": 1630643832229
},
"partitions": [
  {
    "type": "ORM_CV",
    "serializer": "change_vector_json",
    "format": "JSON",
    "payload": {
      "serializerInfo": {
        "id": "3375153340802439707",
        "name": "json_gson",
        "format": "JSON"
      },
      "data": {
        "txId": null,
        "partitionId": null,
        "changeSets": [
          {
            "createEvents": [
              {
                "alias": "sbp.com.sbt.dataspace.jpa.Product",
                "id": "7001023693027540993",
                "primitives": {
                  "isDeleted": false,
                  "partitionId": 0,
                  "lastChangeDate": "2021-08-27T11:27:32.837+03:00",
                  "syalActive": false,
                  "type": "Product"
                },
                "references": {
                  "statusForService": "3"
                },
                "primitiveCollections": {},
                "referenceCollections": {},
                "version": null
              }
            ],
            "updateEvents": [],
            "deleteEvents": [],
            "snapshotEvents": []
          }
        ]
      }
    }
  }
]
}

Версия из поля заголовка rootVersion подставляется в Архивирование (ARCH) во все дочерние ChangeEvent поля, передаваемые в текущем векторе.

Отдельный протокол Init для передачи агрегатов#

Процесс передачи агрегатов#

Для передачи агрегатов используются все те же интерфейсы InitDataSampleApi и InitDataSampleLoad.

InitDataSampleLoad был модифицирован – в него был добавлен метод loadInitMultipart:

@Api
public interface InitDataSampleApi {
    @ApiMethod(apiName = "initLoad", version = "0.0.1")
    String initLoad(String type);
 
    @ApiMethod(apiName = "getBatchCount", version = "0.0.1")
    EstimateResult getBatchCount(String loadingId) throws BatchEstimateException;
 
    @ApiMethod(apiName = "loadBatchAsync", version = "0.0.1")
    void loadBatchAsync(String loadingId, int index, String requestId) throws BatchLoadException;
 
    @ApiMethod(apiName = "abort", version = "0.0.1")
    void abort(String loadingId);
}
 
@Api
public interface InitDataSampleLoad {
    @ApiMethod(apiName = "loadInitBatch", version = "0.0.1")
    void loadInitBatch(InitBatchResult initBatchResult) throws PartitionResultAcceptException;
 
    /*новый метод*/
    @ApiMethod(apiName = "loadInitMultipart", version = "0.0.1")
    void loadInitMultipart(InitMultipartResult initMultipartResult) throws PartitionResultAcceptException;
}

Порядок работы следующий:

1. DataSpace реализует у себя интерфейс InitDataSampleApi, а Архивирование (ARCH) – InitDataSampleLoad. В качестве транспортного слоя используется протокол 4-3. Транспортный протокол kafka

2. Архивирование (ARCH) вызывает у DataSpace метод com.sbt.pprbod.data.transport.init.InitDataSampleApi#initLoad, передавая тип в качестве аргумента. DataSpace возвращает в ответ идентификатор выгрузки (допустимо возвращать в качестве идентификатора выгрузки полное имя класса)

3. Архивирование (ARCH) вызывает у DataSpace метод com.sbt.pprbod.data.transport.init.InitDataSampleApi#getBatchCount передавая в качестве аргумента loadingId, возвращенный в шаге 2. DataSpace возвращает EstimateResult содержащий estimateCode = PS_ADAPTIVE_READY и size=количество агрегатов (то есть количество объектов прогружаемого типа). В случае если процесс оценки длительный, и выполняется в фоновом режиме – DataSpace может вернуть estimateCode = PS_ADAPTIVE_PENDING и size=0. Архивирование (ARCH) в этом случае будет периодически повторно вызывать данный метод до тех пор, пока не получит estimateCode = PS_ADAPTIVE_READY

4. Архивирование (ARCH) последовательно вызывает у DataSpace метод com.sbt.pprbod.data.transport.init.InitDataSampleApi#loadBatchAsync с аргументами:

• loadingId=идентификатор выгрузки из шага 2;

• index=0…size-1;

• requestId=UUID, сгенерированный случайным образом.

5. DataSpace в ответ на вызов loadBatchAsync передает в Архивирование (ARCH) последовательно DataChunk–массивы агрегата, соответствующего индексу из пункта 4b.

   Примечание – В index передается именно порядковый номер агрегата в выгрузке, а не первичный ключ. Нумерация происходит от нуля до size, возвращенного из метода getBatchCount.

   Передача осуществляется путем вызова DataSpace метода com.sbt.pprbod.data.transport.init.InitDataSampleLoad#loadInitMultipart на стороне Архивирование (ARCH) (1 вызов = 1 dataChunk).

   В качестве аргумента передается объект типа InitBatchResult, заполненный по следующим правилам:

   private final long responseId;             // ID ответа (генерируется на стороне DataSapce, рекомендуется чтоб это было монотонно возрастающее числом в разрезе агрегата)
     private final String requestId;            // ID запроса с которым коррелирует ответ (requestId из шага 4 пункт "c")
     private final Timestamp responseTimestamp; // Время формирования dataChunk-а
     private final int  partTotal;              // Количество dataChunk-ов
     private final int  partCurrent;            // Номер текущего dataChunk-а (важно заполнять его в соответствии с порядком применения векторов изменений)
     private final String partHash;             // хеш SHA256 от data
   
     private final String entryType;            // Тип агрегата
     private final String zoneId;               // название шарда. соответствует параметру pprbod.cloud.client.zone из application.properties
     private final byte[] data;                  // java-сериализованный DataContainer (см. раздел ниже)
     private final String hash;                 // не используется, передавать null
   

6. В случае, если выгрузка была отменена пользователем либо на стороне Архивирование (ARCH) произошла ошибка, Архивирование (ARCH) вызывает у DataSpace метод com.sbt.pprbod.data.transport.init.InitDataSampleApi#abort. В реализации этого метода DataSpace может освободить занимаемые ресурсы (если таковые имеются) и считать выгрузку по переданному в качестве аргумента loadingId законченной.

7. В случае, если выгрузка завершена штатно (был успешно загружен путем выполнения шагов 4 и 5 последний агрегат с индексом size-1), abort не вызывается.

Высокоуровневая библиотека для реализации Init агрегатов на стороне DataSpace#

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

Стартер spring-boot устроен таким образом, что потребителю достаточно описать в контексте bean com.sbt.pprbod.data.kafka.InitDataSampleMultipartFacade:

@Component
public class FacadeImpl implements InitDataSampleMultipartFacade {
    /**
     * Здесь выполняется подготовка ресурсов для инициирующей выгрузки
     *
     * @param type тип
     * @return идентификатор выгрузки, либо type, если не требуется какой-либо специальной подготовки ресурсов
     */
    public String initLoad(String type) {
        ...
    }
 
    /**
     * Здесь выполняется подсчет и возврат кол-ва агрегатов по идентификтору выгрузки из {@link #initLoad(String)}
     *
     * @param loadingId идентификатор выгрузки
     * @return результат подсчета
     */
    @Override
    public EstimateResult getBatchCount(String loadingId) {
        ...
    }
 
 
    /**
     * Здесь выполняется загрузка dataChunk-ов агрегата по loadingId и порядковому номеру агрегата в выгрузке
     *
     * @param loadingId идентификатор выгрузки
     * @param index     номер агрегата
     * @return MultipartCursor, реализующий {@link Iterable<com.sbt.pprbod.data.transport.entity.DataContainer>}
     * и дополнительно содержащий метод {@link MultipartCursor#partsCount()} возвращающий количество dataChunk-ов,
     * на которые был разбит агрегат
     */
    @Override
    public MultipartCursor loadBatch(String loadingId, int index) {
        ...
    }
 
    /**
     * Освобождение ресурсов
     *
     * @param loadingId идентификатор выгрузки
     */
    @Override
    public void abort(String loadingId) {
 
    }
}

Правила заполнения DataContainer#

В высокоуровневой реализации нужно предоставить в конструктор функцию, которая будет возвращать Iterable\<DataContainer> (MultipartCursor)

В самом интерфейсе API в методе com.sbt.pprbod.data.transport.init.InitDataSampleLoad#loadInitMultipart нужно отправлять в Архивирование (ARCH) com.sbt.pprbod.data.transport.entity.InitMultipartResult, который в качестве контента содержит java-сериализованный DataContainer.

DataContaner заполняется следующим образом:

/**
 * Контейнер данных
 */
public class DataContainer implements Serializable {
 
    private String key;             //id агрегата (первичный ключ)
    private String entryType;       // тип объекта (полное имя с пакетом)
    private Long version;           // версия агрегата
    private OperationType operType; // тип операции (для Init – всегда create)
    private List<String> updAttrs;  // не используется, пустой список передавать нужно
    private byte[] data;            // utf-8 байты от json-а универсального вектора изменений, содержащего dataChunk

Артефакт#

<dependency>
   <groupId>sbp.ts.pprbod</groupId>
   <artifactId>pprbod-transport-kafka-lib-starter</artifactId>
   <version>04.004.05</version>
</dependency>