ETL API - BatchContext Changes

Given that a dataset could be created at configure time if no fields are provided macros, a check should be available for plugin developers to see whether the dataset already exists at runtime. We can do this by altering the runtime context object passed in to the prepareRun method. As the object extends BatchContext which extends DatasetContext, we can create a new method in BatchContext that checks for the existence of a dataset.

This method will return whether or not the dataset with the provided datasetName already exists and can be used in prepareRun:

If The dataset to write to is macro-substituted and a macro is using in the config, we have to defer dataset creation to prepareRun rather than doing this in the configure stage. Deferring dataset creation in prepareRun will require adding a new method to BatchContext. 

@Beta
public interface BatchContext extends DatasetContext, TransformContext {

/**
 * create dataset identified by datasetName, typeName with properties.
 */
void createDataset(String datasetName, String typeName, DatasetProperties properties);

/**
 * return true if dataset with datasetName exists 
 */
boolean datasetExists(String datasetName);

...
}

@Override
public void prepareRun(BatchSinkContext context) {
  if (!context.datasetExists(config.getName())) {
    pipelineConfigurer.createDataset(config.getName(), ...);
  }
  // ...
}

Notes

Currently if a stream given in stream source or table given in table source doesn't exist, we create a new stream/table. We want to allow table creation as we want to create external dataset for sources, but disallow stream creation, so we are adding only createDataset to the BatchContext.

However there are certain fields which are used to determine the schema in the plugin and those cannot be macro-substituted as schema validation is essential during configure time and we want to disallow macro usage for them. 

Custom Action Setting Config Values:

One use case of the feature is to allow custom actions that run before a plugin to set macros. Custom actions can use workflow tokens to set values for field names. since workflow tokens are merged with runtime arguments and exposed in macro property lookup, macro substitution has access to tokens set from custom action, workflow tokens has higher priority over runtime arguments and preferences.  

Scoping (Low priority):

Scoping is currently at low priority and can be done manually. In our example config from a JDBC source to a table sink, there is a common macro "${table-name}", if the user wants to provide a different name for the table-name in Table Sink, he can manually do this:

This is more of the user creating unique argument keys as opposed to scoping.

Documentation/Changes

Regardless of where the substitution occurs, the guidelines for creating Hydrator plugins would have to change. For existing plugins, any validation for properties that are macro-substitutable in configurePipeline must be moved to prepareRun (see reference section for specific plugins). We also must document the convention for nulling/defaulting macroable properties at configure time.

Implementation Details

interface MacroContext {	
	/**
	 * Given the macro key, return the substituted value
     */ 
	String getValue(String macroKey);
}

Based on the macro type, one of the below MacroContext's will be used to get the value for macro. 
 
DefaultMacroContext implements MacroContext {
	Map<String, String> runtimeArguments;
	String getValue(String macroKey) {
		return runtimeArguments.get(macroKey);
	}
}

SecureMacroContext implements MacroContext {
	SecureStore secureStore;
	String getValue(String macroKey) {
		return secureStore.get(macroKey);
	}
}

RuntimeFunctionMacro implements MacroContext {	
	long logicalStartTime;
	Function<String, String> timezoneFunction;
	String getValue(String arguments) {
		return timezoneFunction.apply(arguments);
	}
}