sort.pm

普通的ETL工具

# vim:ts=4 sw=4
# ----------------------------------------------------------------------------------------------------
#  Name		: ETL::Pequel3::Type::Section::Sort.pm
#  Created	: 22 June 2006
#  Author	: Mario Gaffiero (gaffie)
#
# Copyright 1999-2007 Mario Gaffiero.
# 
# This file is part of Pequel(TM).
# 
# Pequel is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; version 2 of the License.
# 
# Pequel is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License
# along with Pequel; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
# ----------------------------------------------------------------------------------------------------
# Modification History
# When          Version     Who     What
# ----------------------------------------------------------------------------------------------------
package ETL::Pequel3::Type::Section::Sort;
require 5.005_62;
use strict;
use warnings;
use ETL::Pequel3::Type::Section;
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::Section::Sort::Abstract;
	use base qw(ETL::Pequel3::Type::Section::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'target_mem_name', default => 'section_name'),
		Class::STL::ClassMembers::DataMember->new(name => 'item_type', default => 'ETL::Pequel3::Type::Section::Sort::Item'),
		Class::STL::ClassMembers::DataMember->new(name => 'element_type', default => "@{[ __PACKAGE__ ]}");
	use Class::STL::ClassMembers::Constructor;

	package ETL::Pequel3::Type::Section::Sort::Item;
	use base qw(ETL::Pequel3::Type::Section::Item::Abstract);
	use Class::STL::ClassMembers;
	use Class::STL::ClassMembers::Constructor;
	sub new_extra
	{
		my $self = shift;
		$self->attributes()->push_back(
			ETL::Pequel3::Type::Properties::FieldName->new(required => 1),
			ETL::Pequel3::Type::Properties::PequelType->new(),
			ETL::Pequel3::Type::Properties::SortDirection->new()
		);
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::Section::Sort;
	use base qw(ETL::Pequel3::Type::Section::Sort::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'section_name', default => 'sort'),
		Class::STL::ClassMembers::DataMember->new(name => 'element_type', default => "@{[ __PACKAGE__ ]}"),
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'Use this section to perform a sort on the input data stream. Please refer to the specific F<sort_on_input> and F<sort_on_output> section type descriptions for detailed information.');
	use Class::STL::ClassMembers::Constructor;
	sub new_extra
	{
		my $self = shift;
		$self->attributes()->push_back(
			ETL::Pequel3::Type::Properties::On->new(),
		);
		return $self;
	}
	sub select
	{
		my $self = shift;
		my %p = @_;
		$self->err()->user_error(10211, "Section '@{[ 
			$self->section_name() ]}' select() function requires 'on' parameter!")
			unless (exists($p{on}));
		return $self->pequel_ref()->section("@{[ $self->section_name() ]}_on_@{[ lc($p{on}) ]}")->select(@_);
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::Section::SortOnInput;
	use base qw(ETL::Pequel3::Type::Section::Sort::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'section_name', default => 'sort_on_input'),
		Class::STL::ClassMembers::DataMember->new(name => 'element_type', default => "@{[ __PACKAGE__ ]}"),
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'Use this section to perform a sort on the input data stream. One or more input fields can be specified. The F<pequel_type> item attribute is used to specify a I<numeric> or I<string> sort. The F<sort_direction> attribute is used to specify a I<descending> or I<ascending> sort order. The standard Unix I<sort> command is used to perform the sort. If the input data is pre sorted then this section is not required (even if break processing is activated with a F<group by> section declaration). The F<sort_by_on_input> section is not required when the F<hash> option is specified.');
	use Class::STL::ClassMembers::Constructor;
	sub prepare
	{
		my $self = shift;
		$self->SUPER::prepare(@_);
		$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
			. "Sort On Input:"
		);
		my $ds = $self->pequel_ref()->input()->input_dataset();
		$ds->datasource()->sorter(
			$self->pequel_ref()->catalogue()->sorters()->exists($self->pequel_ref()->catalogue()->properties()->sorter())->new(
				uniq => 0,
				field_delimiter => $ds->can('field_delimiter') ? $ds->field_delimiter() : undef,
				pequel_ref => $self->pequel_ref()
			)
		);
		foreach my $f ($self->items()->to_array()) 
		{
			my $pequel_type = $f->pequel_type() || 'string';
			$pequel_type = $self->pequel_ref()->catalogue()->pequel_types()->exists($pequel_type)
				|| $self->err()->user_error(10235, "Unkown pequel-type '@{[ 
					$pequel_type ]}' for sort (on input) field '@{[ $f->field_name() ]}'");

			my $input_field = $self->pequel_ref()->input()->input_fields()->exists($f->field_name()) 
				|| $self->err()->user_error(10236, "Unknown input field '@{[ 
					$f->field_name() ]}' for sort (on input) field '@{[ $f->field_name() ]}'");

			$ds->datasource()->sorter()->add(
				name => $f->field_name(),
				column => $input_field->field_number(),
				direction => $f->sort_direction(),
				pequel_type => $pequel_type,
				pequel_ref => $self->pequel_ref(),
			);
			my $sf = $ds->datasource()->sorter()->fields()->back();
			$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
				. "Sort On Input:"
				. $sf->name()
				. ','
				. $sf->column()
				. ','
				. $sf->direction()
				. ','
				. $sf->pequel_type()->name()
			);
		}
		if ($ds->datasource()->datasource_name() eq 'pequel') 
		{
			my $sub = $self->pequel_ref()->exists($ds->datasource()->target_name())
				|| $self->err()->user_error(10237, "Cannot find input-join sub-pequel @{[ 
					$ds->datasource()->target_name() ]}");
			$sub->output()->output_dataset()->datasource()->sorter($ds->datasource()->sorter());

			$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
				. "Sort On Input --> "
				. $sub->pequel_name()
				. " <-- Sort On Output"
			);
		}
		elsif ($ds->name() eq 'dbi') 
		{
			$ds->select_order_by
			(
				join
				(
					', ', 
					map
					(
						"@{[ $_->name() ]} @{[ $_->direction() ]}", 
						$ds->datasource()->sorter()->fields()->to_array()
					)
				)
			);
			$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
				. "Sort On Input --> "
				. "DBI:" . $ds->datasource()->target_name()
				. " <-- SQL order_by: "
				. $ds->select_order_by()
			);
		}
	}
}
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::Section::SortOnOutput;
	use base qw(ETL::Pequel3::Type::Section::Sort::Abstract);
	use Class::STL::ClassMembers
		Class::STL::ClassMembers::DataMember->new(name => 'section_name', default => 'sort_on_output'),
		Class::STL::ClassMembers::DataMember->new(name => 'element_type', default => "@{[ __PACKAGE__ ]}"),
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => 'Use this section to perform a sort on the output data stream. One or more output fields can be specified. The F<pequel_type> item attribute is used to specify a I<numeric> or I<string> sort. The F<sort_direction> attribute is used to specify a I<descending> or I<ascending> sort order. The standard Unix I<sort> command is used to perform the sort.');
	use Class::STL::ClassMembers::Constructor;
	sub prepare
	{
		my $self = shift;
		$self->SUPER::prepare(@_);
		$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
			. "Sort On Output:"
		);
		# create sorter and plug into output_dataset
		my $ds = $self->pequel_ref()->output()->output_dataset();
		$ds->datasource()->sorter(
			$self->pequel_ref()->catalogue()->sorters()->exists($self->pequel_ref()->catalogue()->properties()->sorter())->new(
				uniq => 0,
				field_delimiter => $ds->can('field_delimiter') ? $ds->field_delimiter() : undef,
				pequel_ref => $self->pequel_ref()
			)
		);
		foreach my $f ($self->items()->to_array()) 
		{
			my $pequel_type = $f->pequel_type() || 'string';
			$pequel_type = $self->pequel_ref()->catalogue()->pequel_types()->exists($pequel_type)
				|| $self->err()->user_error(10238, "Unkown pequel-type '@{[ 
					$pequel_type ]}' for sort (on output) field '@{[ $f->field_name() ]}'");

			my $output_field = $self->pequel_ref()->output()->output_fields()->exists($f->field_name()) 
				|| $self->err()->user_error(10239, "Unknown output field '@{[ 
					$f->field_name() ]}' for sort (on output) field '@{[ $f->field_name() ]}'");

			$ds->datasource()->sorter()->add(
				name => $f->field_name(),
				column => $output_field->field_number(),
				pequel_type => $pequel_type,
				pequel_ref => $self->pequel_ref(),
			);
			my $sf = $ds->datasource()->sorter()->fields()->back();
			$self->err()->trace_msg(10, "@{[ $self->pequel_ref()->pequel_name() ]} -->"
				. "Sort On Output:"
				. $sf->name()
				. ','
				. $sf->column()
				. ','
				. $sf->pequel_type()->name()
			);
		}
	}
}
# ----------------------------------------------------------------------------------------------------
1;