-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: support for templating in table names (#24)
This changeset introduces a simple templating language in the target
`table` configuration.
Previously, there were two options:
1. Unset Table Configuration (=default): The connector used the same QuestDB
table as the topic name from which the message originated. When the connector
was set to listen on multiple topics, it ingested into multiple QuestDB
tables.
2. Explicit Table Configuration: The connector ingested into the one configured
table, regardless of the topic from which the messages originated.
This change allows the use of templates in table configurations.
For example, setting table to `${topic}` will cause ingestion
into the table named after the topic from which the message
originated. This behavior mirrors the scenario where the table
configuration is not set.
However, it also supports more advanced scenarios, such as `${topic}_${key}`,
where key is a string representation of the message key.
Supported Placeholders:
1. `${topic}`
2. `${key}` - string representation of the message key or 'null'
More placeholders may be added in the future.
Caveats:
1. The key is intended for use with simple values and not with complex objects
such as structs or arrays.
2. Using `${key}` can result in a large number of tables. QuestDB might require
tuning when handling thousands of tables.- Loading branch information
Showing
4 changed files
with
248 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| package io.questdb.kafka; | ||
|
|
||
| import io.questdb.std.str.StringSink; | ||
| import org.apache.kafka.connect.errors.ConnectException; | ||
| import org.apache.kafka.connect.sink.SinkRecord; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.function.Function; | ||
|
|
||
| final class Templating { | ||
| private Templating() { | ||
| } | ||
|
|
||
| static Function<SinkRecord, ? extends CharSequence> newTableTableFn(String template) { | ||
| if (template == null || template.isEmpty()) { | ||
| return SinkRecord::topic; | ||
| } | ||
| int currentPos = 0; | ||
| List<Function<SinkRecord, String>> partials = null; | ||
| for (;;) { | ||
| int templateStartPos = template.indexOf("${", currentPos); | ||
| if (templateStartPos == -1) { | ||
| break; | ||
| } | ||
| int templateEndPos = template.indexOf('}', templateStartPos + 2); | ||
| if (templateEndPos == -1) { | ||
| throw new ConnectException("Unbalanced brackets in a table template, missing closing '}', table template: '" + template + "'"); | ||
| } | ||
| int nextTemplateStartPos = template.indexOf("${", templateStartPos + 1); | ||
| if (nextTemplateStartPos != -1 && nextTemplateStartPos < templateEndPos) { | ||
| throw new ConnectException("Nesting templates in a table name are not supported, table template: '" + template + "'"); | ||
| } | ||
| String templateName = template.substring(templateStartPos + 2, templateEndPos); | ||
| if (templateName.isEmpty()) { | ||
| throw new ConnectException("Empty template in table name, table template: '" + template + "'"); | ||
| } | ||
| if (partials == null) { | ||
| partials = new ArrayList<>(); | ||
| } | ||
| String literal = template.substring(currentPos, templateStartPos); | ||
| if (!literal.isEmpty()) { | ||
| partials.add(record -> literal); | ||
| } | ||
| switch (templateName) { | ||
| case "topic": { | ||
| partials.add(SinkRecord::topic); | ||
| break; | ||
| } | ||
| case "key": { | ||
| partials.add(record -> record.key() == null ? "null" : record.key().toString()); | ||
| break; | ||
| } | ||
| default: { | ||
| throw new ConnectException("Unknown template in table name, table template: '" + template + "'"); | ||
| } | ||
| } | ||
| currentPos = templateEndPos + 1; | ||
| } | ||
| if (partials == null) { | ||
| return record -> template; | ||
| } | ||
| String literal = template.substring(currentPos); | ||
| if (!literal.isEmpty()) { | ||
| partials.add(record -> literal); | ||
| } | ||
| List<Function<SinkRecord, String>> finalPartials = partials; | ||
| StringSink sink = new StringSink(); | ||
| return record -> { | ||
| sink.clear(); | ||
| for (Function<SinkRecord, String> partial : finalPartials) { | ||
| sink.put(partial.apply(record)); | ||
| } | ||
| return sink; | ||
| }; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
105 changes: 105 additions & 0 deletions
105
connector/src/test/java/io/questdb/kafka/TemplatingTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| package io.questdb.kafka; | ||
|
|
||
| import org.apache.kafka.connect.errors.ConnectException; | ||
| import org.apache.kafka.connect.sink.SinkRecord; | ||
| import org.junit.Assert; | ||
| import org.junit.Test; | ||
|
|
||
| import java.util.function.Function; | ||
|
|
||
| public class TemplatingTest { | ||
|
|
||
| @Test | ||
| public void testPlainTableName() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("table"); | ||
| SinkRecord record = newSinkRecord("topic", "key"); | ||
| assertTableName(fn, record, "table"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testEmptyTableName() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn(""); | ||
| SinkRecord record = newSinkRecord("topic", "key"); | ||
| assertTableName(fn, record, "topic"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testNullTableName() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn(null); | ||
| SinkRecord record = newSinkRecord("topic", "key"); | ||
| assertTableName(fn, record, "topic"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testSimpleTopicTemplate() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("${topic}"); | ||
| SinkRecord record = newSinkRecord("mytopic", "key"); | ||
| assertTableName(fn, record, "mytopic"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testTopicWithKeyTemplates() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("${topic}_${key}"); | ||
| SinkRecord record = newSinkRecord("mytopic", "mykey"); | ||
| assertTableName(fn, record, "mytopic_mykey"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testTopicWithNullKey() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("${topic}_${key}"); | ||
| SinkRecord record = newSinkRecord("mytopic", null); | ||
| assertTableName(fn, record, "mytopic_null"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testMissingClosingBrackets() { | ||
| assertIllegalTemplate("${topic", "Unbalanced brackets in a table template, missing closing '}', table template: '${topic'"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testOverlappingTemplates() { | ||
| assertIllegalTemplate("${topic${key}", "Nesting templates in a table name are not supported, table template: '${topic${key}'"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testEmptyTemplate() { | ||
| assertIllegalTemplate("${}", "Empty template in table name, table template: '${}'"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testIllegalTemplate() { | ||
| assertIllegalTemplate("${unknown}", "Unknown template in table name, table template: '${unknown}'"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testSuffixLiteral() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("${topic}_suffix"); | ||
| SinkRecord record = newSinkRecord("mytopic", "key"); | ||
| assertTableName(fn, record, "mytopic_suffix"); | ||
| } | ||
|
|
||
| private static void assertIllegalTemplate(String template, String expectedMessage) { | ||
| try { | ||
| Templating.newTableTableFn(template); | ||
| Assert.fail(); | ||
| } catch (ConnectException e) { | ||
| Assert.assertEquals(expectedMessage, e.getMessage()); | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| public void testTopicWithEmptyKey() { | ||
| Function<SinkRecord, ? extends CharSequence> fn = Templating.newTableTableFn("${topic}_${key}"); | ||
| SinkRecord record = newSinkRecord("mytopic", ""); | ||
| assertTableName(fn, record, "mytopic_"); | ||
| } | ||
|
|
||
| private static void assertTableName(Function<SinkRecord, ? extends CharSequence> fn, SinkRecord record, String expectedTable) { | ||
| Assert.assertEquals(expectedTable, fn.apply(record).toString()); | ||
| } | ||
|
|
||
| private static SinkRecord newSinkRecord(String topic, String key) { | ||
| return new SinkRecord(topic, 0, null, key, null, null, 0); | ||
| } | ||
|
|
||
| } |