Feat/kafka mapping support

src/Plugin/Queue/Handlers/Source/BaseCreateSourceHandler.php

@@ -67,7 +67,7 @@ protected static function checkAndCreateSource(Client $manticoreClient): void {
 		$sql = /** @lang ManticoreSearch */
 			'CREATE TABLE ' . Payload::SOURCE_TABLE_NAME .
 			' (id bigint, type text, name text attribute indexed, '.
-			'full_name text, buffer_table text, attrs json, original_query text)';
+			'full_name text, buffer_table text, attrs json, custom_mapping json, original_query text)';
 
 		$request = $manticoreClient->sendRequest($sql);
 		if ($request->hasError()) {

src/Plugin/Queue/Handlers/Source/CreateKafka.php

@@ -13,7 +13,9 @@
 
 use Manticoresearch\Buddy\Base\Plugin\Queue\Handlers\View\CreateViewHandler;
 use Manticoresearch\Buddy\Base\Plugin\Queue\Payload;
+use Manticoresearch\Buddy\Core\Error\GenericError;
 use Manticoresearch\Buddy\Core\Error\ManticoreSearchClientError;
+use Manticoresearch\Buddy\Core\Error\QueryValidationError;
 use Manticoresearch\Buddy\Core\ManticoreSearch\Client;
 use Manticoresearch\Buddy\Core\Task\TaskResult;
 use Manticoresearch\Buddy\Core\Tool\Buddy;
@@ -160,12 +162,13 @@ public static function handle(Payload $payload, Client $manticoreClient): TaskRe
 			}
 
 			$escapedPayload = str_replace("'", "\\'", $payload->originQuery);
+			$customMapping = str_replace("'", "\\'", $options->customMapping);
 
 			$query = /** @lang ManticoreSearch */
 				'INSERT INTO ' . Payload::SOURCE_TABLE_NAME .
-				' (id, type, name, full_name, buffer_table, attrs, original_query) VALUES ' .
+				' (id, type, name, full_name, buffer_table, attrs, custom_mapping, original_query) VALUES ' .
 				"(0, '" . self::SOURCE_TYPE_KAFKA . "', '$options->name','{$options->name}_$i'," .
-				"'{$bufferTablePrefix}{$options->name}_$i', '$attrs', '$escapedPayload')";
+				"'{$bufferTablePrefix}{$options->name}_$i', '$attrs', '$customMapping', '$escapedPayload')";
 
 			$request = $manticoreClient->sendRequest($query);
 			if ($request->hasError()) {
@@ -317,7 +320,10 @@ public static function parseOptions(Payload $payload): \stdClass {
 
 		$parsedPayload = $payload->model->getPayload();
 		$result->name = strtolower($parsedPayload['SOURCE']['name']);
-		$result->schema = strtolower($parsedPayload['SOURCE']['create-def']['base_expr']);
+
+		$mapping = self::parseMapping($parsedPayload['SOURCE']['create-def']['sub_tree']);
+		$result->customMapping = $mapping['customMapping'];
+		$result->schema = $mapping['schema'];
 
 		foreach ($parsedPayload['SOURCE']['options'] as $option) {
 			if (!isset($option['sub_tree'][0]['base_expr'])) {
@@ -340,4 +346,48 @@ public static function parseOptions(Payload $payload): \stdClass {
 		}
 		return $result;
 	}
+
+	/**
+   * @param array{
+	 *                      expr_type: string,
+	 *                      base_expr: string,
+	 *                      sub_tree: array{
+	 *                          expr_type: string,
+	 *                          base_expr: string,
+	 *                          sub_tree: array{
+	 *                              expr_type: string,
+	 *                              base_expr: string
+	 *                          }[]
+	 *                      }[]
+	 *                  }[] $fields
+   *
+   * @return array{customMapping: non-empty-string|false, schema:non-falsy-string}
+	 * @throws GenericError
+*/
+	public static function parseMapping(array $fields): array {
+		$schema = [];
+		$customMapping = [];
+		$pattern = '/^`?([a-zA-Z0-9_]+)`?\s*[\'"]+(.*)[\'"]+\s([a-zA-Z]+)$/usi';
+
+		foreach ($fields as $field) {
+			$definition = strtolower($field['base_expr']);
+
+			if (preg_match($pattern, $definition, $matches)) {
+				$schema[] = $matches[1].' '.$matches[3];
+				$customMapping[$matches[1]] = $matches[2];
+				continue;
+			}
+
+			$schema[] = $definition	;
+		}
+
+		$encodedMapping = json_encode($customMapping);
+		if ($encodedMapping === false) {
+			QueryValidationError::throw('Incorrect custom mapping provided');
+		}
+		return [
+			'customMapping' => $encodedMapping,
+			'schema' => '('.implode(',', $schema).')',
+		];
+	}
 }

src/Plugin/Queue/Handlers/Source/DropSourceHandler.php

@@ -104,6 +104,12 @@ protected static function removeSourceRowData(array $sourceRow, Client $client):
 		foreach ($queries as $query) {
 			$request = $client->sendRequest($query);
 			if ($request->hasError()) {
+				if (str_contains(
+					(string)$request->getError(),
+					"unknown table '".Payload::VIEWS_TABLE_NAME."' in update request"
+				)) {
+					continue;
+				}
 				throw ManticoreSearchClientError::create((string)$request->getError());
 			}
 		}

src/Plugin/Queue/QueueProcess.php

@@ -104,6 +104,7 @@ public function runPool(): void {
 	 *    buffer_table:string,
 	 *    destination_name:string,
 	 *    query:string,
+	 *    custom_mapping: string,
 	 *    attrs:string } $instance
 	 * @param bool $shouldStart
 	 * @throws \Exception

src/Plugin/Queue/Workers/Kafka/KafkaWorker.php

@@ -28,7 +28,9 @@ class KafkaWorker implements WorkerRunnerInterface
 
 	private Client $client;
 	private string $brokerList;
-	/** @var array|string[] */
+	/** @var array<string, string> */
+	private array $customMapping;
+	/** @var array<int, string> */
 	private array $topicList;
 	private string $consumerGroup;
 	private string $bufferTable;
@@ -45,6 +47,7 @@ class KafkaWorker implements WorkerRunnerInterface
 	 *   full_name:string,
 	 *   buffer_table:string,
 	 *   destination_name:string,
+	 *   custom_mapping: string,
 	 *   query:string,
 	 *   attrs:string } $instance
 	 *
@@ -61,6 +64,15 @@ public function __construct(
 		$this->client = $client;
 		$this->consumerGroup = $attrs['group'];
 		$this->brokerList = $attrs['broker'];
+
+		$decodedMapping = json_decode($instance['custom_mapping'], true);
+		if ($decodedMapping === false) {
+			GenericError::throw(
+				'Custom mapping decoding error: '.json_last_error_msg()
+			);
+		}
+		/** @var array<string,string> $decodedMapping */
+		$this->customMapping = $decodedMapping;
 		$this->bufferTable = $instance['buffer_table'];
 		$this->topicList = array_map(fn($item) => trim($item), explode(',', $attrs['topic']));
 		$this->batchSize = (int)$attrs['batch'];
@@ -208,8 +220,12 @@ private function mapMessages(array $batch): array {
 	private function handleRow(array $message): array {
 		$row = [];
 		foreach ($this->fields as $fieldName => $fieldType) {
-			if (isset($message[$fieldName])) {
-				$row[$fieldName] = $this->morphValuesByFieldType($message[$fieldName], $fieldType);
+			$inputKeyName = $fieldName;
+			if (isset($this->customMapping[$fieldName])) {
+				$inputKeyName = $this->customMapping[$fieldName];
+			}
+			if (isset($message[$inputKeyName])) {
+				$row[$fieldName] = $this->morphValuesByFieldType($message[$inputKeyName], $fieldType);
 			} else {
 				if (in_array(
 					$fieldType, [Fields::TYPE_INT,