inputjoin.pm

普通的ETL工具

# vim:ts=4 sw=4
# ----------------------------------------------------------------------------------------------------
#  Name		: ETL::Pequel3::Type::Section::InputJoin.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::InputJoin;
require 5.005_62;
use strict;
use warnings;
use ETL::Pequel3::Type::Section;
# ----------------------------------------------------------------------------------------------------
{
	package ETL::Pequel3::Type::Section::InputJoin;
	use base qw(ETL::Pequel3::Type::Section::Abstract);
	use Class::STL::ClassMembers
		qw( 
			pequel_name
			config
			join_name
			input_dataset 
			input_fields 
			key_field 
			key_field_ref
			join_field
			join_field_ref
			key_type 
			transfer 
			transfer_dataset 
		),
		Class::STL::ClassMembers::DataMember->new(name => 'target_mem_name', default => 'join_name'),
		Class::STL::ClassMembers::DataMember->new(name => 'section_name', default => 'input_join'),
		Class::STL::ClassMembers::DataMember->new(name => 'item_type', 
			default => 'ETL::Pequel3::Type::Section::InputJoin::Item'),
		Class::STL::ClassMembers::DataMember->new(name => 'element_type', default => "@{[ __PACKAGE__ ]}"),
		Class::STL::ClassMembers::DataMember->new(name => 'description', default => "With F<input_join> multiple inputs can be merged in together with the main F<input>. All F<input_join> datasets and the main F<input> dataset must have the same sort order on the F<join_field>/F<key_field>. Currently only I<left-outer-join> is supported -- one record for each matching join key from the F<input_join> is returned. Any F<input> records without a matching F<join_field>/F<key_field> in the F<input_join> will have null values for the F<input_join> fields returned.\n\n\nAn F<input_join> should be used instead of a table whenever possible -- unlike tables, they are not memory intensive, and they do not require any preload, so there will be no wait for the actual tranformation process to begin.\n\n\nThe data for an F<input_join> is accessed similarly to a table but without the F<key> argument: access the data for C<FIELD_1> in F<input_join> C<J1> with C<%J1->E<gt>C<FIELD_1>.");
	use Class::STL::ClassMembers::Constructor;
	sub new_extra
	{
		my $self = shift;
		$self->attributes()->push_back(
			$self->attributes()->factory(name => 'join_name', required => 1),
			ETL::Pequel3::Type::Properties::DataSetSpec->new(required => 1), 
			ETL::Pequel3::Type::Properties::Transfer->new(),
			$self->attributes()->factory(name => 'key_field', required => 1),
			$self->attributes()->factory(name => 'join_field', required => 1),
			$self->attributes()->factory(name => 'key_type'),
		);
		$self->config($self->pequel_ref()->config())
			if (defined($self->pequel_ref())); #NOTE: need this because pequel_ref is self below;
	}
	sub select
	{
		my $self = shift;
		my %p = @_;
		$self->err()->user_error(10211, "Section '@{[ 
			$self->section_name() ]}' select() function requires '@{[ $self->target_mem_name() ]}' parameter!")
			unless (exists($p{$self->target_mem_name()}));
		my $s;
		return $s if (($s = $self->exists($p{$self->target_mem_name()})) != 0);
		$self->add(%p);
		$self->back()->attributes()->set(@_);
		return $self->back();
	}
	sub prepare
	{
		my $self = shift;
		$self->config($self->pequel_ref()->config())
			if (defined($self->pequel_ref())); 
		$self->pequel_name($self->join_name()); #??? unless (defined($self->pequel_name()));
		$self->err()->user_error(10221, "Attribute @{[ __PACKAGE__ ]}::configis undefined!")
			unless (defined($self->config()));
		$self->err()->user_error(10226, "Unknown input field name '@{[ 
			$self->join_field() ]}' specified for join-field!")
			unless ($self->pequel_ref()->input()->input_fields()->exists($self->join_field()));

		$self->join_field_ref($self->pequel_ref()->input()->input_fields()->exists($self->join_field()));
		$self->err()->user_error(10227, "Join-field '@{[ 
			$self->join_field() ]}' must not be an input derived field!")
			if (defined($self->join_field_ref()->calc_exp()));

		$self->input_dataset($self->pequel_ref()->factories()->input_datasets()->factory(
			dataset_spec => $self->attributes()->dataset_spec(), 
			fdname => '_J_' . $self->join_name(),
#<			configuration => $self->configuration(),
			pequel_ref => $self->pequel_ref(),
		));
		$self->input_fields($self->pequel_ref()->catalogue()->fieldsets()->input_join_fields()->new(
			dataset => $self->input_dataset(), 
			pequel_ref => $self, # NOTE: fields are owned by (this) specific join;
		));
		$self->input_dataset()->input_fields($self->input_fields());

		$self->input_fields()->map_input($self->items());
		$self->input_dataset()->datasource()->sorter(
			$self->pequel_ref()->catalogue()->sorters()->exists($self->pequel_ref()->properties()->sorter())->new(
				uniq => 0,
				field_delimiter => $self->input_dataset()->can('field_delimiter') 
					? $self->input_dataset()->field_delimiter() : undef,
				pequel_ref => $self->pequel_ref()
			)
		);
		$self->input_dataset()->map_output(
			main => $self->pequel_ref(),
			key_field => $self->key_field(), 
			sorter => $self->input_dataset()->datasource()->sorter(), # plug into input_dataset
		);

		$self->err()->user_error(10232, "Empty input-field list for input join '@{[ $self->join_name() ]}'!")
			unless ($self->input_fields()->size());
		$self->err()->user_error(10233, "Cannot find key-field '@{[ 
			$self->key_field() ]}' in data-source @{[ $self->input_dataset()->datasource()->target_name() ]}")
			unless ($self->input_fields()->exists($self->key_field()));

		$self->key_field_ref($self->input_fields()->exists($self->key_field()));
		$self->err()->user_error(10230, "Join key-field @{[ $self->name() ]}->@{[ $self->key_field() 
				]} must not be a derived field!")
			if (defined($self->key_field_ref()->calc_exp()));

		$self->input_dataset()->datasource()->sorter()->add(
			name => $self->key_field(),
			column => $self->key_field_ref()->field_number(),
			pequel_type => $self->key_field_ref()->pequel_type(),
			pequel_ref => $self->pequel_ref()
		);

		$self->pequel_ref()->output()->transfer_fields()->map_input_fields(
			$self->input_fields(), 
			$self->transfer_dataset($self->pequel_ref()->factories()->output_datasets()->factory( 
				dataset_spec => 'void',
				fdname => '_O_' . $self->join_name(),
				pequel_ref => $self->pequel_ref(),
			)),
			($self->pequel_ref()->group_by()->input_fields()->size() ? $self->transfer() : 'last'),
			$self->join_name(),
		)
		if ($self->transfer());
	}
	sub parse
	{
		my $self = shift;
		my $parser = ETL::Pequel3::Parser->new(
			user_joins => $self->pequel_ref()->user_joins(),
			pequel_ref  => $self->pequel_ref(),
			macro_use_list => $self->pequel_ref()->macro_use_list(),
			translate_fields => ETL::Pequel3::Parser::TranslateFields::Input->new( #-->translate_fieldnames
				input_fields => $self->input_fields())
		);
		foreach (grep(defined($_->calc_exp()), $self->input_fields()->to_array())) 
		{
			$_->calc_exp_raw($parser->used_by($_)->translate($_->calc_exp())->raw());
		}
	}
	sub translate
	{
		my $self = shift;
		my $field = shift; #< || undef;
		my $show_synonyms = shift; #TODO: try target_obj->[pequel_ref()->]configuration()->show_synonym();
		my $target_obj = shift; #TODO: not best solution?
		$self->err()->user_error(10302, "Undefined field for input-join '@{[ 
			$self->join_name() ]}'->translate() function")
			unless (defined($field));
		my $c = ETL::Pequel3::CodeStyler::Program::Perl->new();
		if (!ref($field)) 
		{
			$field =~ s/^@{[ $self->join_name() ]}->//;
			my $f = $self->input_fields()->exists($field); 
			$self->err()->user_error(10302, "Unknown field '$field' for input-join '@{[ 
				$self->join_name() ]}'")
				unless (defined($f) && ref($f));
			$field = $f;
		}
		$c->code($field->getvar());
		$target_obj->xref()->references($field) if ($target_obj->can('xref'));
		$field->xref()->referenced_by($target_obj) if ($field->can('xref'));
		return $c;
	}

	package ETL::Pequel3::Type::Section::InputJoin::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::PequelType->new(required => 0),
			ETL::Pequel3::Type::Properties::FieldName->new(required => 1),
			$self->attributes()->factory(name => 'operator'),
			$self->attributes()->factory(name => 'calc_exp'),
			$self->attributes()->factory(name => 'xml_type', default => 'PCDATA', validate => '^(attr|PCDATA|CDATA)$'),
		);
	}
}
# ----------------------------------------------------------------------------------------------------
1;