[feature][functions]Create a built-in Function implementing the most common basic transformations #15903

cbornet · 2022-06-02T15:25:42Z

Motivation

This PR implements PIP-173 : Create a built-in Function implementing the most common basic transformations.

It provides a first set of basic functions:

cast: modifies the key or value schema to a target compatible schema passed in the schema-type argument. This PR only enables STRING schema-type. The part argument allows to choose on which part to apply between key and value. If part is null or absent the transformations applies to both the key and value.
drop-fields: drops fields given as a string list in parameter fields. The part argument allows to choose on which part to apply between key and value. If part is null or absent the transformations aplpies to both the key and value. Currently only AVRO is supported.
merge-key-value: merges the fields of KeyValue records where both the key and value are structured types of the same schema type. Currently only AVRO is supported.
unwrap-key-value: if the record is a KeyValue, extract the KeyValue's value and make it the record value. If parameter unwrapKey is present and set to true, extract the KeyValue's key instead.

Modifications

Add a transforms module in multi-module pulsar-function project.
Create classes TransformFunction, TransformContext, interface TransformStep described in PIP-173
Create implementations CastStep, DropFieldsStep, MergeKeyValueStep, UnwrapKeyValueStep with tests
Add logic to parse userConfig in TransformFunction and call the TransformStep implementations.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

Run tests from the transforms module

Does this pull request potentially affect one of the following parts:

no

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API: (yes / no)
The schema: (yes / no / don't know)
The default values of configurations: (yes / no)
The wire protocol: (yes / no)
The rest endpoints: (yes / no)
The admin cli options: (yes / no)
Anything that affects deployment: (yes / no / don't know)

Documentation

Check the box below or label this PR directly.

Need to update docs?

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

eolivelli · 2022-06-06T10:04:23Z

pulsar-functions/transforms/src/main/java/org/apache/pulsar/functions/transforms/CastStep.java

+import org.apache.pulsar.common.schema.SchemaType;
+
+@Slf4j
+public class CastStep implements TransformStep {


the purpose if this Step is not clear to me.
what does this step do ?

It changes the schema to another compatible one. The name is inspired from SMTs

eolivelli · 2022-06-06T10:05:50Z

...tions/transforms/src/main/java/org/apache/pulsar/functions/transforms/MergeKeyValueStep.java

+    @Override
+    public void process(TransformContext transformContext) {
+        Schema<?> keySchema = transformContext.getKeySchema();
+        if (keySchema == null) {


what does it mean ?
are we assuming that the Key is byte[] or String when we are not using KeyValue Schema ?

It means that if we don't have a KeyValue, we don't do anything.

eolivelli · 2022-06-06T10:06:36Z

...ctions/transforms/src/main/java/org/apache/pulsar/functions/transforms/TransformContext.java

+        this.context = context;
+        this.outputTopic = context.getOutputTopic();
+        Schema<?> schema = currentRecord.getSchema();
+        //TODO: should we make a copy ?


no need to perform a copy, but maybe it is better to wrap to Collections.unmodifiableMap() if we are not supposed to change the contents

That's the thing. We want to give the possibility of steps to modify the properties. Wrapping in an unmodifiable map is a good idea as the step would have to make the copy itself.

I changed to lazily set the properties before sending the message if it was not set by one of the steps.
Note that PulsarRecord properties are already unmodifiable.
This can be changed when we implement a step that modifies the properties.

eolivelli · 2022-06-06T10:06:49Z

...ctions/transforms/src/main/java/org/apache/pulsar/functions/transforms/TransformContext.java

+    private KeyValueEncodingType keyValueEncodingType;
+    private String key;
+    private Map<String, String> properties;
+    private String outputTopic;


Steps should have the possibility to modify the output topic. So we can implement basic routing operations such as RegexRouter

eolivelli · 2022-06-06T10:06:53Z

...ctions/transforms/src/main/java/org/apache/pulsar/functions/transforms/TransformContext.java

+@Slf4j
+@Data
+public class TransformContext {
+    private Context context;


eolivelli · 2022-06-06T10:07:51Z

...ctions/transforms/src/main/java/org/apache/pulsar/functions/transforms/TransformContext.java

+        TypedMessageBuilder<?> message = context.newOutputMessage(outputTopic, outputSchema)
+                .properties(properties)
+                .value(outputObject);
+        if (key != null) {


if we are using KeyValue Schema we should not set the Key this way.

eolivelli · 2022-06-06T10:08:01Z

...tions/transforms/src/main/java/org/apache/pulsar/functions/transforms/TransformFunction.java

+import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.api.Record;
+
+@Slf4j


please add javadoc

eolivelli · 2022-06-06T10:09:30Z

pulsar-functions/transforms/src/main/java/org/apache/pulsar/functions/transforms/CastStep.java

+            if (keySchemaType == SchemaType.STRING) {
+                outputSchema = Schema.STRING;
+                outputKeyObject = outputKeyObject.toString();
+            }


else ?
we should throw a IllegalStateException here if we are not supporting other types

eolivelli · 2022-06-06T10:09:34Z

pulsar-functions/transforms/src/main/java/org/apache/pulsar/functions/transforms/CastStep.java

+            transformContext.setKeySchema(outputSchema);
+            transformContext.setKeyObject(outputKeyObject);
+        }
+        if (valueSchemaType == SchemaType.STRING) {


else ?
we should throw a IllegalStateException here if we are not supporting other types

eolivelli · 2022-06-06T10:10:48Z

...ions/transforms/src/main/java/org/apache/pulsar/functions/transforms/UnwrapKeyValueStep.java

+                transformContext.setValueSchema(transformContext.getKeySchema());
+                transformContext.setValueObject(transformContext.getKeyObject());
+            }
+            // TODO: can we avoid making the conversion to NATIVE_AVRO ?


can you please clarify ?

Well theoretically we have the Schema.AVRO schema and value of the wrapped key or value and could use that instead of unwrapping the AVRO schema and rewrapping in Schema.NATIVE_AVRO. But there's no clean way to give this information to the TransformContext. So I'll remove this comment for now.

* Reject invalid schema types in cast transfo * Add javadoc * Lazily set message properties

github-actions · 2022-07-15T02:27:50Z

The pr had no activity for 30 days, mark with Stale label.

eolivelli · 2022-08-17T08:25:03Z

Moved to a dedicated repository:
https://github.com/datastax/pulsar-transformations

cbornet added 17 commits June 2, 2022 17:04

Add transform function

fda8a0f

Update function following review

ea01150

Add support for dropping key fields and list of fields

2a731f6

Add Context and outputTopic to TransformRecord

26132d8

Some refactoring

e8c11d4

Add RemoveFieldFunctionTest

dc48ae2

Add tests for Avro KeyValue

65c2abb

Add test for KeyValue with primitives

237b50d

Add TransformFunctionTest

e54d6dc

Add merge and cast to string functions

071bb14

Add unwrap-key-value function

afa4951

Add Licences

551e0bc

Fix wrapping of unmodified KeyValue

7789ae2

Fix tests

d9eab06

Add more tests

0fcd507

Only provide TransformFunction as a function entry point

aad4aba

Add a cache of schemas in MergeKeyValueStep

00491a7

github-actions bot assigned cbornet Jun 2, 2022

github-actions bot added the doc-required Your PR changes impact docs and you will update later. label Jun 2, 2022

Remove wrongly added import

8bcbf69

cbornet changed the title ~~Transforms core~~ Create a built-in Function implementing the most common basic transformations Jun 2, 2022

cbornet changed the title ~~Create a built-in Function implementing the most common basic transformations~~ [improve][functions]Create a built-in Function implementing the most common basic transformations Jun 2, 2022

cbornet changed the title ~~[improve][functions]Create a built-in Function implementing the most common basic transformations~~ [feature][functions]Create a built-in Function implementing the most common basic transformations Jun 2, 2022

Change POM parent

4db3f00

eolivelli requested changes Jun 6, 2022

View reviewed changes

Improve following review:

2ab3d4d

* Reject invalid schema types in cast transfo * Add javadoc * Lazily set message properties

cbornet requested a review from eolivelli June 14, 2022 18:25

github-actions bot added the Stale label Jul 15, 2022

eolivelli closed this Aug 17, 2022

cbornet deleted the transforms-core branch August 17, 2022 16:43

cbornet restored the transforms-core branch August 17, 2022 16:43

cbornet deleted the transforms-core branch August 17, 2022 16:43

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feature][functions]Create a built-in Function implementing the most common basic transformations #15903

[feature][functions]Create a built-in Function implementing the most common basic transformations #15903

cbornet commented Jun 2, 2022 •

edited

Loading

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 9, 2022

eolivelli Jun 6, 2022

cbornet Jun 9, 2022

github-actions bot commented Jul 15, 2022

eolivelli commented Aug 17, 2022

[feature][functions]Create a built-in Function implementing the most common basic transformations #15903

[feature][functions]Create a built-in Function implementing the most common basic transformations #15903

Conversation

cbornet commented Jun 2, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jul 15, 2022

eolivelli commented Aug 17, 2022

cbornet commented Jun 2, 2022 •

edited

Loading