[INLONG-10117][SDK] Support to transform from PB protocol to CSV/KV protocol by single SQL

[luchunliang]

[SDK] Support to transform from PB protocol to CSV/KV protocol by single SQL

Fixes #10117

Parent issue:
#10022

Motivation

Support to transform from PB protocol to CSV/KV protocol by single SQL

Modifications

• Support to parse the following PB protocol
  syntax = "proto3";
  package test;
  message SdkMessage {
  bytes msg = 1;
  int64 msgTime = 2;
  map<string, string> extinfo = 3;
  }
  message SdkDataRequest {
  string sid = 1;
  repeated SdkMessage msgs = 2;
  uint64 packageID = 3;
  }

• Support to parse the following PB data
  SdkDataRequest.Builder requestBuilder = SdkDataRequest.newBuilder();
  requestBuilder.setSid("sid");
  requestBuilder.setPackageID(1);
  {
  SdkMessage.Builder msgBuilder = SdkMessage.newBuilder();
  msgBuilder.setMsgTime(1713243918000L);
  msgBuilder.setMsg(ByteString.copyFrom("msgValue4".getBytes()));
  msgBuilder.putExtinfo("key", "value");
  SdkMessage msgObj = msgBuilder.build();
  requestBuilder.addMsgs(msgObj);
  }
  {
  SdkMessage.Builder msgBuilder = SdkMessage.newBuilder();
  msgBuilder.setMsgTime(1713243918002L);
  msgBuilder.setMsg(ByteString.copyFrom("msgValue42".getBytes()));
  msgBuilder.putExtinfo("key2", "value2");
  SdkMessage msgObj = msgBuilder.build();
  requestBuilder.addMsgs(msgObj);
  }
  SdkDataRequest requestObj = requestBuilder.build();
  byte[] srcBytes = requestObj.toByteArray();

• Generate the final data by the following SQL
  select $root.sid,
  $root.packageID,
  $child.msgTime,
  $child.msg from source

• The final data is the following string
  sid|1|1713243918000|msgValue4
  sid|1|1713243918002|msgValue42

Verifying this change

(Please pick either of the following options)

• This change is a trivial rework/code cleanup without any test coverage.

• This change is already covered by existing tests, such as:
  (please describe tests)

• This change added tests and can be verified as follows:

  (example:)

  • Added integration tests for end-to-end deployment with large payloads (10MB)
  • Extended integration test for recovery after broker failure

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation