Transform Action

Description

A Transform Action transforms the content it receives. It may also update metadata and add annotations.

Java

Interface

A TransformAction must implement the transform method which receives:

• ActionContext describing the action's environment and current execution
• ActionParameters containing flow parameters specified for the action
• TransformInput providing the content and metadata to be transformed

Transform Input

public class TransformInput {
    List<ActionContent> content;
    Map<String, String> metadata;
}

Return Types

The transform method must return a TransformResultType, which is implemented by TransformResult, TransformResults, ErrorResult, and FilterResult.

The TransformResult contains the content, metadata, and annotations created by the TransformAction. The TransformResults contains a list of ChildTransformResult. Each ChildTransformResult creates a new child DeltaFile that continues through the same flow.

Example

package org.deltafi.example;

import org.deltafi.actionkit.action.transform.TransformAction;
import org.deltafi.actionkit.action.transform.TransformInput;
import org.deltafi.actionkit.action.transform.TransformResult;
import org.deltafi.actionkit.action.transform.TransformResultType;
import org.deltafi.common.types.ActionContext;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;

@Component
public class HelloWorldTransformAction extends TransformAction<Parameters> {
    public HelloWorldTransformAction() {
        super("Add some content noting that we did a really good job");
    }

    @Override
    public TransformResultType transform(@NotNull ActionContext context, @NotNull Parameters params, @NotNull TransformInput input) {
        if ("true".equals(input.getMetadata().getOrDefault("filter", ""))) {
            return new FilterResult(context, "Filter due to filter metadata");
        }
        if ("true".equals(input.getMetadata().getOrDefault("error", ""))) {
            return new ErrorResult(context, "Error due to error metadata");
        }
        
        String data = input.content(0).loadString() + "\nHelloWorldTransformAction did a great job";

        TransformResult result = new TransformResult(context);
        result.addMetadata("transformKey", "transformValue");
        result.addAnnotation("transformAnnotation", "value");
        result.saveContent(data, "transform-named-me", "text/plain");
        return result;
    }
}

Python

Interface

A TransformAction must implement the transform method which receives:

• Context describing the action's environment and current execution
• BaseModel containing flow parameters for use by the action, matching the type specified by the param_class() method, which must inherit from BaseMmodel, or a default/empty BaseModel if unspecified.
• TransformInput providing the content and metadata to be transformed

Transform Input

class TransformInput(NamedTuple):
    content: List[Content]
    metadata: dict

Return Types

The transform() method must return one of: TransformResult, TransformResults, ErrorResult, or FilterResult.

The TransformResult contains the content, metadata, and annotations created by the TransformAction. The TransformResults contains a list of ChildTransformResult. Each ChildTransformResult creates a new child DeltaFile that continues through the same flow.

Example

from deltafi.action import TransformAction
from deltafi.domain import Context
from deltafi.input import TransformInput
from deltafi.result import ErrorResult, FilterResult, TransformResult
from pydantic import BaseModel


class HelloWorldTransformAction(TransformAction):
    def __init__(self):
        super().__init__('Add some content noting that we did a really good job')

    def transform(self, context: Context, params: BaseModel, transform_input: TransformInput):
        context.logger.info(f"Transforming {context.did}")
        if transform_input.metadata.get('filter') == 'true':
            return FilterResult(context, 'Filter due to filter metadata')
        if transform_input.metadata.get('error') == 'true':
            return ErrorResult(context, 'Error due to error metadata')

        data = f"{transform_input.content[0].load_str()}\nHelloWorldTransformAction did a great job"

        return TransformResult(context)
                .save_string_content(data, 'transform-named-me', 'text/plain')
                .add_metadata('transformKey', 'transformValue')
                .annotate('transformAnnotation', 'value')

Go

Beta

The Go action kit is currently in beta. Interfaces may change in future releases.

Interface

A TransformAction must implement the Transform method which receives:

• *Deltafile providing the input content, metadata, and context fields
• A typed params struct containing flow parameters specified for the action

The action returns the same *Deltafile, which accumulates output content, metadata changes, and annotations.

Deltafile

The Deltafile is the primary type action authors interact with. It carries context fields (DID, Name, FlowName, etc.), provides access to input content and metadata, and accumulates output content and metadata changes.

// Context fields (read-only)
df.Did            // DeltaFile ID
df.Name           // original filename
df.DataSource     // data source name
df.FlowName       // flow name
df.ActionName     // action name

// Input content
df.FirstContent()           // *ActionContent, error
df.ContentAt(index)         // *ActionContent, error
df.ContentNamed(name)       // *ActionContent, error
df.GetContentList()        // []ActionContent
df.HasContent()             // bool

// Output content
df.SaveStringContent(name, mediaType, data)   // *ActionContent, error
df.SaveBytesContent(name, mediaType, data)    // *ActionContent, error
df.AddContent(content)                         // *Deltafile (chainable)
df.PassthroughContent()                        // *Deltafile (chainable)

// Child DeltaFiles
df.AddChild(childName)        // *Deltafile (chainable)

// Metadata
df.GetMetadata(key)                  // string, error
df.GetMetadataOrDefault(key, def)    // string
df.AddMetadata("key", "value")       // *Deltafile (chainable)
df.DeleteMetadataKeys("key1")        // *Deltafile (chainable)

// Annotations
df.AddAnnotation("key", "value")     // *Deltafile (chainable)

// Error/Filter signaling
df.Errorf("format", args...)         // *Deltafile (chainable)
df.Filterf("format", args...)        // *Deltafile (chainable)

Return Value

The Transform method always returns a *Deltafile. The kit inspects the returned Deltafile to determine the outcome:

• Errorf was called — the DeltaFile is errored.
• Filterf was called — the DeltaFile is filtered.
• One or more children were added via AddChild — each child becomes a new DeltaFile that continues through the flow.
• Otherwise, the action succeeds, and any saved content, metadata changes, and annotations are applied to the DeltaFile.

File layout

Each Go action lives in two files inside the plugin's actions/ directory:

• <name>_action.go — the action struct, init() registration, and the action method (Transform, Egress, etc.)
• <name>_action_params.go — the typed *Params struct only

Splitting params into a sibling file keeps the action code focused on logic and gives the schema round-trip tooling (described below) a deterministic location to read from and write to.

Parameter struct tags

The kit reads the following struct tags when generating JSON Schema:

Tag	Effect
json:"name"	parameter name displayed in the DeltaFi UI
description:"..."	human-readable description
default:"..."	default value (string form; coerced to the field type)
enum:"a,b,c"	allowed values (renders as a dropdown)
required:"true"	marks the parameter as required
minimum:"0" / maximum:"100"	numeric bounds

Supported field types: Go primitives (bool, int, string, float64), slices, string-keyed maps, pointers to other structs declared in the same params file, and the kit's actionkit.DataSize and actionkit.EnvVar helpers.

Schema round-trip tooling

The kit ships a cmd/actionkit-schemagen binary, invoked from every plugin's Makefile:

make jsonFromParams   # writes actions/<name>_action_params.json from each _params.go
make paramsFromJson   # writes actions/<name>_action_params.go   from each _params.json

jsonFromParams produces schemas that are byte-for-byte identical to what the kit sends to core during plugin registration (the AST tool reuses the same actionkit.GenerateSchema pathway). paramsFromJson regenerates the Go source from a schema; the output is canonical (fields alphabetize, comments drop, slice-of-struct fields come back as slice-of-pointer-to-struct) but functionally equivalent.

Example

actions/hello_world_go_transform_action.go:

package actions

import (
    actionkit "gitlab.com/deltafi/deltafi/deltafi-go-action-kit/v2"
)

func init() {
    actionkit.RegisterTransform("HelloWorldGoTransformAction", &HelloWorldGoTransformAction{}).
        WithParams(HelloWorldGoTransformParams{}).
        Describe("Add some content noting that we did a really good job")
}

type HelloWorldGoTransformAction struct{}

func (a *HelloWorldGoTransformAction) Transform(df *actionkit.Deltafile, params HelloWorldGoTransformParams) *actionkit.Deltafile {
    if df.GetMetadataOrDefault("filter", "") == "true" {
        return df.Filterf("Filter due to filter metadata")
    }
    if df.GetMetadataOrDefault("error", "") == "true" {
        return df.Errorf("Error due to error metadata")
    }

    content, err := df.FirstContent()
    if err != nil {
        return df.Errorf("Failed to load content").SetErrorContext(err.Error())
    }

    data, err := content.LoadString()
    if err != nil {
        return df.Errorf("Failed to load content").SetErrorContext(err.Error())
    }

    transformed := data + "\nHelloWorldGoTransformAction did a great job"

    if _, err := df.SaveStringContent("transform-named-me", "text/plain", transformed); err != nil {
        return df.Errorf("Failed to save content").SetErrorContext(err.Error())
    }

    df.AddMetadata("transformKey", "transformValue")
    df.AddAnnotation("transformAnnotation", "value")
    return df
}

actions/hello_world_go_transform_action_params.go:

package actions

// HelloWorldGoTransformParams defines the action's configurable parameters.
type HelloWorldGoTransformParams struct {
    // Add fields with json + description/default tags to expose them in the UI.
}

C++

Beta

The C++ action kit is currently in beta. Interfaces may change in future releases.

Interface

A TransformAction must implement the transform method which receives:

• A typed Params& containing flow parameters specified for the action, when the action declares a Params type
• deltafi::DeltaFile& providing the input content, metadata, and context fields

The action returns the same deltafi::DeltaFile&, which accumulates output content, metadata changes, and annotations. An action with parameters implements transform(Params& params, deltafi::DeltaFile& df); an action without them implements transform(deltafi::DeltaFile& df).

DeltaFile

The DeltaFile is the primary type action authors interact with. It carries context fields (DID, name, flow name, etc.), provides access to input content and metadata, and accumulates output content and metadata changes.

// Context fields (read-only)
df.did();            // DeltaFile ID
df.name();           // DeltaFile name, typically the original filename
df.data_source();    // data source name
df.flow_name();      // flow name
df.action_name();    // action name
df.context();        // ActionContext&: lookup and artifact clients, hostname, start time, join details

// Input content
df.first_content();          // ActionContent&; throws ExpectedContentError when there is no content
df.content_at(0);            // ActionContent&; throws ExpectedContentError when out of range
df.content_named("in.txt");  // ActionContent*; nullptr when absent
df.content();                // const std::vector<ActionContent>&
df.has_content();            // bool

// Output content
df.save_content("data", "out.txt", "text/plain");  // ActionContent&; also accepts std::vector<uint8_t>
df.save_content_from_writer(                       // ActionContent&; streams in bounded memory
    [](std::ostream& out) { out << "data"; }, "out.txt", "text/plain");
df.add_content(df.first_content());                // add existing content without saving it again
df.passthrough_content();                          // add all input content

// Child DeltaFiles
df.add_child("child-name");  // DeltaFile&

// Metadata
df.get_metadata("key");                   // std::string; throws MissingMetadataError when absent
df.get_metadata_or_default("key", "def");  // std::string
df.add_metadata("key", "value");
df.delete_metadata_key("key");

// Annotations
df.add_annotation("key", "value");

// Logging and metrics
df.log_info("message");
df.log_info() << "streamed " << 42 << " values";
df.add_metric(deltafi::Metric{.name = "records", .value = 42, .tags = {}});

// Error/Filter signaling
df.error("cause", "optional context");     // DeltaFile&
df.filter("message", "optional context");  // DeltaFile&

The ActionContent& returned by save_content and save_content_from_writer stays valid for the life of the DeltaFile: later saves, add_content, passthrough_content, and add_child never invalidate it.

Return Value

The transform method always returns the DeltaFile& it received. The kit inspects it to determine the outcome:

• error was called — the DeltaFile is errored.
• filter was called — the DeltaFile is filtered.
• One or more children were added via add_child — each child becomes a new DeltaFile that continues through the flow. A child carries content, metadata changes, annotations, log messages, and metrics. Adding content, metadata changes, or annotations to the parent as well, or calling error, filter, or add_child on a child, errors the DeltaFile with a cause naming the misuse.
• Otherwise, the action succeeds, and any saved content, metadata changes, and annotations are applied to the DeltaFile.

The action may instead throw deltafi::DeltaFileErrorException or deltafi::DeltaFileFilterException to error or filter the DeltaFile. Any other std::exception errors the DeltaFile with the exception message as the cause, and anything else thrown errors it with the cause "unknown error". Content saved by an execution that errors or filters is removed from storage.

Example

actions/hello_world_cpp_transform_action.hpp:

#pragma once

#include <algorithm>
#include <cctype>
#include <string>

#include <deltafi/plugin>

struct HelloWorldCppTransformActionParams {
    bool convert_to_upper = false;

    DELTAFI_PARAMS(HelloWorldCppTransformActionParams,
        FIELD(convert_to_upper, "If true, convert the transformed content to uppercase"))
};

class HelloWorldCppTransformAction {
public:
    using Params = HelloWorldCppTransformActionParams;

    deltafi::DeltaFile& transform(Params& params, deltafi::DeltaFile& df) {
        if (df.get_metadata_or_default("filter", "") == "true") {
            return df.filter("Filter due to filter metadata");
        }
        if (df.get_metadata_or_default("error", "") == "true") {
            return df.error("Error due to error metadata");
        }
        if (!df.has_content()) {
            return df.error("No content provided");
        }

        auto data = df.first_content().load_string() + "\nHelloWorldCppTransformAction did a great job";
        if (params.convert_to_upper) {
            std::ranges::transform(data, data.begin(),
                [](unsigned char c) { return static_cast<char>(std::toupper(c)); });
        }

        df.save_content(data, "transform-named-me", "text/plain");
        df.add_metadata("transformKey", "transformValue");
        df.add_annotation("transformAnnotation", "value");
        return df;
    }
};

DELTAFI_ACTION(HelloWorldCppTransformAction,
    "HelloWorldCppTransformAction",
    "Add some content noting that we did a really good job")

DELTAFI_PARAMS generates the JSON Schema the plugin registers for the parameters, converting snake_case field names to camelCase (convert_to_upper becomes convertToUpper). See Action Parameters for nested structs and EnvVar.