RDF::Query::Plan::ThresholdUnion - Executable query plan for unions.

 Gregory Todd Williams <gwilliams@cpan.org>

# RDF::Query::Plan::ThresholdUnion
# -----------------------------------------------------------------------------

package RDF::Query::Plan::ThresholdUnion;

use strict;
use warnings;
use base qw(RDF::Query::Plan);

use Time::HiRes qw(time);
use Scalar::Util qw(blessed);

use RDF::Query::ExecutionContext;

######################################################################

our ($VERSION);
BEGIN {
	$VERSION	= '2.907';
}

######################################################################

sub new {
	my $class	= shift;
	my $time	= shift;
	my @plans	= @_;
	my $self	= $class->SUPER::new( $time, \@plans );
	my %vars;
	foreach my $plan (@plans) {
		foreach my $v ($plan->referenced_variables) {
			$vars{ $v }++;
		}
	}
	$self->[0]{referenced_variables}	= [ keys %vars ];
	return $self;
}

sub execute ($) {
	my $self	= shift;
	my $context	= shift;
	if ($self->state == $self->OPEN) {
		throw RDF::Query::Error::ExecutionError -text => "ThresholdUnion plan can't be executed while already open";
	}

	my $l		= Log::Log4perl->get_logger("rdf.query.plan.thresholdunion");
	
	my $iter	= $self->[2][0];
	$l->trace("threshold union initialized with first sub-plan: " . $iter->sse);
	
	$iter->execute( $context );
	
	if ($iter->state == $self->OPEN) {
		$self->[0]{iter}		= $iter;
		$self->[0]{idx}			= 0;
		$self->[0]{context}		= $context;
		$self->[0]{start_time}	= time;
		$self->state( $self->OPEN );
	} else {
		warn "no iterator in execute()";
	}
	$self;
}

sub next {
	my $self	= shift;
	unless ($self->state == $self->OPEN) {
		throw RDF::Query::Error::ExecutionError -text => "next() cannot be called on an un-open ThresholdUnion";
	}
	my $iter	= $self->[0]{iter};
	my $row		= $iter->next;
	my $l		= Log::Log4perl->get_logger("rdf.query.plan.thresholdunion");
	if ($row) {
		return $row;
	} else {
		$l->trace("thresholdunion sub-plan finished");
		delete $self->[0]{iter};
		return undef unless ($self->[0]{idx} < $#{ $self->[2] });
		$iter->close();
		
		my $iter;
		my $index;
		my $threshold	= $self->threshold_time;
		my $elapsed		= time - $self->[0]{start_time};
		if (($threshold == 0) or ($elapsed < $threshold)) {
			# we haven't passed the threshold of execution time, so go on
			# to the next optimistic plan:
			$l->trace("the elapsed time ($elapsed) hasn't passed the threshold time ($threshold); continuing to next plan");
			$index	= ++$self->[0]{idx};
			$iter	= $self->[2][ $index ];
		} else {
			# the elapsed time has passed the threshold time, so jump straight to the default plan
			$l->trace("the elapsed time ($elapsed) has passed the threshold time ($threshold); jumping to the default plan");
			$index	= ($self->[0]{idx} = $#{ $self->[2] });
			$iter	= $self->[2][ $index ];
		}
		
		$l->trace("threshold union executing next sub-plan: " . $iter->sse);
		$iter->execute( $self->[0]{context} );
		if ($iter->state == $self->OPEN) {
			$self->[0]{iter}	= $iter;
			return $self->next;
		} else {
			throw RDF::Query::Error::ExecutionError -text => "execute() on child [${index}] of UNION failed during next()";
		}
	}
}

sub close {
	my $self	= shift;
	unless ($self->state == $self->OPEN) {
		throw RDF::Query::Error::ExecutionError -text => "close() cannot be called on an un-open ThresholdUnion";
	}
	if ($self->[0]{iter}) {
		$self->[0]{iter}->close();
		delete $self->[0]{iter};
	}
	$self->SUPER::close();
}

sub children {
	my $self	= shift;
	return @{ $self->[2] };
}

sub threshold_time {
	my $self	= shift;
	return $self->[1];
}

sub optimistic {
	my $self	= shift;
	return @{ $self->[2] }[ 0 .. $#{ $self->[2] } - 1 ];
}

sub default {
	my $self	= shift;
	return $self->[2][ $#{ $self->[2] } ];
}

sub distinct {
	return 0;
}

sub ordered {
	return [];
}

sub plan_node_name {
	return 'threshold-union';
}

sub plan_prototype {
	my $self	= shift;
	return qw(i *P);
}

sub plan_node_data {
	my $self	= shift;
	my $exprs	= $self->[2];
	return ($self->threshold_time, $self->children);
}

sub graph {
	my $self	= shift;
	my $g		= shift;
	my (@children)	= map { $_->graph( $g ) } ($self->children);
	$g->add_node( "$self", label => "Threshold Union" . $self->graph_labels );
	$g->add_edge( "$self", $_ ) for (@children);
	return "$self";
}

1;

__END__