Perl language aggregates

package Maxae;
use nzae::Ae;
use strict;
use autodie;
our @ISA = qw(nzae::Ae);
my $ae = Maxae->new();
$ae->run();
# The runUda function here is the exact function implemented for the Ae class
# Reproducing this function is not needed in user code to run an aggregate
# as it is inherited. The user can however override this function to have
# finer control on the running of the aggregate Ae
sub runUda
{
my $self = shift;
while(1)
{
my $aggregationType = $self->_getNextAggregation();
if ( $aggregationType == $self->getAggregateTypeInitialize() )
{
$self->_initializeState();
$self->_saveAggregateResult();
}
elsif ( $aggregationType == $self->getAggregateTypeAccumulate() )
{
$self->_accumulate($self->getState(), $self->getInputRow());
$self->_saveAggregateResult();
}
elsif ( $aggregationType == $self->getAggregateTypeMerge() )
{
$self->_merge($self->getState(), $self->getInputState());
$self->_saveAggregateResult();
}
elsif ( $aggregationType == $self->getAggregateTypeFinalResult() )
{
my $result = $self->_finalResult($self->getState());
$self->_setAggregateResult($result, 1);
$self->_saveAggregateResult();
}
elsif ( $aggregationType == $self->getAggregateTypeEnd() )
{
return;
}
elsif ( $aggregationType == $self->getAggregateTypeError() )
{
croak(nzae::Exceptions::AeInternalError->new("Error calling
nzaeAggNext(). Cause unknown."));
}
else
{
croak(nzae::Exceptions::AeInternalError->new("Received unknown
aggregation type."));
}
}
}
#the functions below need to be overridden by the user
#while writing an aggregate Ae
sub _initializeState
{
my $self = shift;
$self->_setState(0, -2147483647);
}
sub _accumulate
{
my $self = shift;
my @instate = shift;
my @row = shift;
my $state = pop(@instate);
if (scalar(@row) > 0)
{
if ( defined $row[0] )
{
unless (defined $state)
{
if (defined $row[0])
{
$state = $row[0];
}
}
elsif ( $state < $row[0] )
{
if (defined $row[0])
{
$state = $row[0];
}
}
}
}
unless (defined $state)
{
$state = -2147483647;
}
$self->_setState(0, $state);
}
sub _merge
{
my $self = shift;
my @instate = shift;
my @inputValues = shift;
my $state = pop(@instate);
for ( my $i = 0 ; $i < scalar(@inputValues); $i++ )
{
unless (defined $state)
{
$state = $inputValues[$i];
}
elsif ( $state < $inputValues[$i] )
{
$state = $inputValues[$i];
}
}
unless (defined $state)
{
$state = -2147483647;
}
$self->_setState(0, $state);
}
sub _finalResult
{
my $self = shift;
my @state = shift;
my $ret = defined $state[0]? $state[0] : -2147483647;
return $ret;
}
1;

In summary, to write a UDA without customization, the steps are:

1. Create a Perl module file. In the example above this is MaxAe.pm.
2. Import and instantiate the nzae::Ae class in the file.
3. Import autodie to handle unhandled exceptions during the course of execution.
4. Override the _initializeState function to initialize the state variable defined in the registration step.
5. Override the accumulate method to implement aggregation functionality for each dataslice.
6. Override the merge method to implement aggregation of state variables from each dataslice.
7. Override the finalResult method to return the merged state variable.
8. Execute the run() method of the nzae::Ae object.
9. Since this is a Perl module, the file should contain a “1” at the end of the file MaxAe.pm.

Deploy the script:

$NZ_EXPORT_DIR/ae/utilities/bin/compile_ae --language perl --version 3 \
--template deploy Maxae.pm;

Register the example as a UDA by using the state parameter:

$NZ_EXPORT_DIR/ae/utilities/bin/register_ae --language perl --version 3 \
--template uda --exe Maxae.pm --sig "maxaepl(int)" --return int4 \
--state "(int4)"

To run, first create a dummy table and then run the aggregate:

CREATE TABLE grp_test (grp int4, val int4);
CREATE TABLE
INSERT INTO grp_test VALUES (1, 1);
INSERT 0 1
INSERT INTO grp_test VALUES (1, 2);
INSERT 0 1
INSERT INTO grp_test VALUES (1, 3);
INSERT 0 1
INSERT INTO grp_test VALUES (2, 4);
INSERT 0 1
SELECT maxaepl(val) FROM grp_test;
MAXAEPL
---------
4
(1 row)
SELECT grp, maxaepl(val) FROM grp_test GROUP BY grp;
GRP | MAXAEPL
-----+---------
2 | 4
1 | 3
(2 rows)
SELECT grp, maxaepl(val) over (partition BY grp) FROM grp_test;
GRP | MAXAEPL
-----+---------
1 | 3
1 | 3
1 | 3
2 | 4
(4 rows)