Filter commands

Filter commands receive input records and produce output records. A filter command may take in one record and output 0-n records or accept 1-n records and output 0-1 records, depending on the intention of the specific command.

The following diagram shows an illustration of filter commands.

Sorting, merging and filtering records

This section describes about the sorting, merging and filtering records.

Sort

The sort parameter sorts records into order specified by the evaluated key, as listed in the following table.

If the number of records to sort is less than the configured buffer-size then this will perform an in-memory sort (the quickest option).

Otherwise, this command perform a k-way merge sort and compress sorted chunks when they are not required.

The following table lists the sort templates in detail.

Working of the sort command

The merge sort from file inputs should deplete a single channel of inputs before getting more records from the source file to avoid unnecessary seeks.

The current implementation has a limitation that it does not persist data to disk for large sizes. However, to mitigate this, data is compressed in chunks and only accessed as required, so memory usage required is expected to be 100- 500 MB for a few million objects. In a complex pipeline this could add up.

Sort culture

SortComparison.OrdinalIgnoreCase is currently hard coded since this sorts hyphens, whereas many cultures, including en-US and en-UK, ignore these, which is likely lead to incorrect sort results.

Merge

The Merge parameter combines consecutive objects with the same key into a single object.

The following table lists the merger parameters in detail.

Merge only considers consecutive records with the same key result, so will frequently follow a Sort command.

The following table lists the merge templates.

Sort and merge multiple sources

The following read multiple sources, defines a sorting of records so that last update (from the second source) wins and either appends or deletes values based on an action column:

Example:

<pipeline>

<read>

<waypoint folder="{{ $[waypoint-root] }}\captured\{{ $interface }}" />

<pin>

<metadata name="sort-order" value="1!!" />

</pin>

</read>

<read template="geicl/InfoInterface/Data Sources/{{ $interface }}">

<file folder="{{ $[source-data] }}\data load" report-progress="true">

<text template="tab" />

</file>

<pin>

<metadata name="id" value="{{ @[tag_number] }}" />

<metadata name="sort-order" value="2!!{{ $[index].padnumbers(9) }}" />

<metadata name="changed" value="y" />

</pin>

</read>

<sort key="{{ $[this.id] }}!!{{ $[this.sort-order] }}" />

<merge

key="{{ $[this.id] }}"

action="{{ if @action in ['1','delete','deleted','del'] then 'DeleteValue' else 'AppendValue' }}"

/>

</pipeline>

Notes:

• Merge with the Data Model design in this document has a potential flaw that if the evaluated merge key encompasses many items, this could break the idea of managed CPU/ memory usage provided via blocking threads with a max buffer size. However, this is very likely to be a configuration flaw and not an intended scenario for this mechanism in the current design.

• When merging via a complex expression, the Registers Gateway may apply a KeyReduce optimisation to avoid unnecessary string comparison effort.

Where

The Where parameter allows simple to complex conditions to be expressed to filter records, as listed in following table.

Filter out records not matching condition

Example:

<where condition="{{ 'Z-WI1' in [ $[this.id], @TAG] or

$[this.index].between(100000, 100500) }}" />

Only returns records where either the metadata id or tag number is 'Z-WI1' or the row index is in the specified range.

<where condition="{{ @[ATTRIBUTE_NAME] not in ['class','unit'] }}" />

Filters out the specified attributes (it could be used when reading a file with an attribute name column).

Projecting records

A field is either spawned From the previous command filter, or defined new with Name and no From.

If From is specified, a new Name may still be given. When From is specified, variables $[attr.name] and $[attr.value] are made available for use in evaluated expressions.

The following table lists the select projection parameters.

The following table lists the SelectFieldsConfig parameters.

If an empty <fields /> tag is added, this will include any fields which have not yet been emitted by other Fields definitions (going by the field name).

The following table lists the SelectFieldConfig parameters.

Map attributes across multiple records to a single record

Select allows mapping from an attribute file format to object. This may then be used, for example, with a subsequent Merge command to combine several object records into a single object record with all the attributes.

Project object attributes across multiple records into a single record per object

For example, if an incoming file has columns TAG_NUMBER, ATTIBUTE_NAME, ATTRIBUTE_VALUE, ATTIRBUTE_UOM, you can map this to objects as follows:

Example:

<pipeline>

<select>

<fields>

<field

name="{{ @[ATTRIBUTE_NAME] }}"

value="{{ @[ATTRIBUTE_VALUE] }}"

units="{{ @[ATTRIBUTE_UOM] }}"

/>

</fields>

<pin>

<metadata name="id" value="{{ @[TAG_NUMBER] }}" />

</pin>

</select>

<merge key="{{ $[this.id] }}" />

</pipeline>

Project new fields and retain anything else

This next example will select the six specified fields plus any other fields being passed into the filter, but will also filter out any dynamic fields which duplicate the 6 specified, so is a little different from an SQL SELECT a,b,c,*:

Example:

<select>

<fields>

<field name="SysID" value="{{ $[this.waypoint.index] }}" />

<field name="ObjectID" value="{{ $[this.id].slice('|', -1) }}" />

<field name="ObjectFullName" value="{{ '|'.join($project, $[this.id]) }}" />

<field name="ObjectDescription" value="{{ $[this.name] }}" />

<field name="ObjectClass" value="{{ COALESCE($[this.class], 'UNKNOWN') }}" />

<field name="ObjectState" value="{{ @[TAG_STATUS] }}" />

</fields>

<fields />

</select>

Modify

The modify parameter modifies the attributes within each processed object with 1-n sub instructions. These updates are evaluated for each value of each attribute of an object. Each instruction has a where condition restricting it.

All modify operations accept a where condition, which defaults to true (that is, apply to all attributes and values).

The following table lists modify projection parameters.

All modify operations may specify these parameters.

The following table lists AttributeUpdateConfig parameters in detail.

The different AttributeUpdateConfig operation types are described below.

Update

The update attribute values operation will update attribute values (one or more of Value, Units, and Dataset), as listed in the following table.

Set

The set attribute operation sets an attribute (create or overwrite), as listed in the following table.

Merge

The merge operation merges values where the where condition is true.

Delete

The delete operation deletes values where the where condition is true.

Apply several attribute modification rules

The following example applies the sanitize function on all attributes and updates values and units where they are set to non-applicable values until they meet the other noted conditions.

Example:

<modify>

<update value="{{ $[attr.value].sanitise() }}" />

<update value=""

where="{{ $[attr.value] in ['N/A', 'V-N/A'] and $[attr.dataset] not equals 'master' }}" />

<update units=""

where="{{ $[attr.units] in ['N/A', 'V-N/A'] and $[this.id] not contains '9002' }}" />

</modify>

Merge attributes according to rule

The following example merges attribute values that have been collected together, ensuring the override dataset wins over others, and set the dataset value for all attributes to Master. This may be used as logic to create a master dataset from multiple source datasets.

Example:

<modify>

<merge where="{{ $[attr.dataset] not equals 'override' }}" />

<merge where="{{ $[attr.dataset] equals 'override' }}" />

<update dataset="Master" />

</modify>

Interpret

The interpret command is used to map the attributes of each object in a stream to a class library. The expectation is that we will want to achieve one of the following results:

• Map to a valid class library artifact (such as an Attribute) and be able to guarantee as much as possible that our data will load into a target system.

• Remove the offending artifact, and report on this.

• Mark an object so that it may be subsequently, reported as an error, or used to error a load. Marking it (a constructive rather than destructive act) allows this decision making to be performed downstream. This will strip out attributes that don't match to the class library (immediate need) and issue a warning count for each attribute striped out.

This differs in nature from the Modify command as it specifically expects to map to the active class library and failure to do so should result in a fail action. Currently, this just allows mapping of an attribute to the class library.

The following table lists the interpret projection parameters.

The following table lists the interpret attribute parameters.

The following table lists the interpret expression parameters.

This command can create available results so that they can be pinned as metadata (or set metadata automatically) and then this could be consumed by later filters (for example, publish 'rejected objects' info to a reporting database and then remove it with a where filter).

Pivot

The pivot command maps all attribute values of an object into distinct object records. If an attribute is unset or has no values, that attribute will not produce any records.

The following table lists the details about pivot projection parameters.

The following table lists the details about PivotFieldConfig parameters.

The following table lists the details about pivot variables available for pinning metadata.

Pivot a record into several attribute records

The following pivots each object into a 0-N records, each with 6 fields, filtering out some attributes. Each record will inherit the metadata from the original object.

Example:

<pivot>

<field name="ObjectID" value="{{ $[this.waypoint.index] }}" />

<field name="ObjectFullName" value="{{ '|'.join($project, $[this.id]) }}" />

<field name="AttributeID" value="{{ $[attr.name] }}" />

<field name="AttributeValue" value="{{ $[attr.value] }}" />

<field name="AttributeUOM" value="{{ $[attr.units] }}" />

<field name="Include" value="{{ if (FieldIsReference.isFalse or FieldPreservesAttribute) and ($[attr.name] not in ['class', 'unit']) then 'y'}}" />

</pivot>

<where condition="{{ @include equals 'y'}}" />

Note: It may or may not make sense to incorporate this into the SELECT command.