Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/kafka mapping support #420

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
+83 −12
Open

Feat/kafka mapping support #420

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
cd72cc4
Sketches of mapping support
Dec 12, 2024
21db304
Merge branch 'main' into feat/kafka-mapping-support
Dec 13, 2024
d2d5970
Merge branch 'main' into feat/kafka-mapping-support
Dec 26, 2024
2c57c80
Sketches of mapping support
Dec 26, 2024
749451c
Review fixes
Dec 26, 2024
afcfa1c
Review fixes
Jan 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions composer.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion src/Plugin/Queue/Handlers/Source/BaseCreateSourceHandler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()) {
Expand Down
57 changes: 54 additions & 3 deletions src/Plugin/Queue/Handlers/Source/CreateKafka.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -129,6 +131,7 @@ public static function handle(Payload $payload, Client $manticoreClient): TaskRe

$options = self::parseOptions($payload);


$sql = /** @lang ManticoreSearch */
'SELECT * FROM ' . Payload::SOURCE_TABLE_NAME .
" WHERE match('@name \"" . $options->name . "\"')";
Expand Down Expand Up @@ -159,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'," .
"'_buffer_{$options->name}_$i', '$attrs', '$escapedPayload')";
"'_buffer_{$options->name}_$i', '$attrs', '$customMapping', '$escapedPayload')";

$request = $manticoreClient->sendRequest($query);
if ($request->hasError()) {
Expand Down Expand Up @@ -316,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'])) {
Expand All @@ -339,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).')',
];
}
}
9 changes: 8 additions & 1 deletion src/Plugin/Queue/Handlers/Source/DropSourceHandler.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,14 +95,21 @@ protected static function removeSourceRowData(array $sourceRow, Client $client):
/** @lang Manticore */
"DROP TABLE {$sourceRow['buffer_table']}",
/** @lang Manticore */
"UPDATE _views SET suspended=1 WHERE match('@source_name \"{$sourceRow['full_name']}\"')",
'UPDATE '.Payload::VIEWS_TABLE_NAME.' SET suspended=1 '.
"WHERE match('@source_name \"{$sourceRow['full_name']}\"')",
/** @lang Manticore */
"DELETE FROM _sources WHERE id = {$sourceRow['id']}",
];

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());
}
}
Expand Down
1 change: 1 addition & 0 deletions src/Plugin/Queue/QueueProcess.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ public function runPool(): void {
* buffer_table:string,
* destination_name:string,
* query:string,
* custom_mapping: string,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We request string here but in parseMapping method returns false in case json failed to encode, is it normal?

Copy link
Contributor Author

@djklim87 djklim87 Dec 26, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have never encountered issues with json_encode. In my opinion, it’s nearly impossible to encode simple array with an error

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

JSON easy to fail in case we accept user data. Does the encoded array get information from the outside of the world? User provide it? in case easy to trigger false so with unsupported Unicodes

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree that adding some additional checks to the mapping provided by the user is a good idea. However, after these checks, we should assume the data is correct and that there is no possibility of encoding/decoding errors.

* attrs:string } $instance
* @param bool $shouldStart
* @throws \Exception
Expand Down
18 changes: 15 additions & 3 deletions src/Plugin/Queue/Workers/Kafka/KafkaWorker.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@ class KafkaWorker implements WorkerRunnerInterface

private Client $client;
private string $brokerList;
/** @var array|string[] */
/** @var array<string, string> */
private array $customMapping;
djklim87 marked this conversation as resolved.
Show resolved Hide resolved
/** @var array<int, string> */
private array $topicList;
private string $consumerGroup;
private string $bufferTable;
Expand All @@ -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
*
Expand All @@ -61,6 +64,10 @@ public function __construct(
$this->client = $client;
$this->consumerGroup = $attrs['group'];
$this->brokerList = $attrs['broker'];

/** @var array<string,string> $decodedMapping */
$decodedMapping = json_decode($instance['custom_mapping'], true);
$this->customMapping = $decodedMapping;
$this->bufferTable = $instance['buffer_table'];
$this->topicList = array_map(fn($item) => trim($item), explode(',', $attrs['topic']));
$this->batchSize = (int)$attrs['batch'];
Expand Down Expand Up @@ -208,8 +215,13 @@ 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 ($this->customMapping !== []
&& isset($this->customMapping[$fieldName])) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isset should be enough?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In case isset is enough let's remove $this->customMapping !== [] cuz it does not do anything, isset is enough

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, because during the construct, we use json_decode. If the JSON is corrupted, it can return false. That’s why we check if it’s an array.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You have declaration

private array $customMapping;

In case it returns false and we try to assign, we will fail already before this code, so I guess [] can be removed

	$decodedMapping = json_decode($instance['custom_mapping'], true);
	$this->customMapping = $decodedMapping;

What if we will do this thing:

	$decodedMapping = json_decode($instance['custom_mapping'], true) ?: []

So it will make sure we have array even if it returns false and avoid strict type fatal error

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m not sure this is a good idea:
1. If we have a corrupted custom mapping, it’s more rational to throw an exception rather than proceed.
2. In most cases, we don’t have any mapping, so checking $this->customMapping !== [] allows us to avoid consuming resources and performing unnecessary isset checks since we already know there’s no mapping.

$inputKeyName = $this->customMapping[$fieldName];
}
if (isset($message[$inputKeyName])) {
$row[$fieldName] = $this->morphValuesByFieldType($message[$inputKeyName], $fieldType);
} else {
if (in_array(
$fieldType, [Fields::TYPE_INT,
Expand Down