@@ -64,6 +64,28 @@ sub referenced_blanks {
return uniq(@list);
}
+=item C<< referenced_functions >>
+
+Returns a list of the Function URIs used in this algebra expression.
+
+=cut
+
+sub referenced_functions {
+ my $self = shift;
+ my @list;
+ foreach my $arg ($self->construct_args) {
+ if (blessed($arg)) {
+ if ($arg->isa('RDF::Query::Expression::Function')) {
+ push(@list, $arg->uri);
+ } elsif ($arg->isa('RDF::Query::Algebra')) {
+ my @funcs = $arg->referenced_functions;
+ push(@list, @funcs);
+ }
+ }
+ }
+ return uniq(@list);
+}
+
=item C<< check_duplicate_blanks >>
Returns true if blank nodes respect the SPARQL rule of no blank-label re-use
RDF-Query/lib/RDF/Query/Algebra.pm
@@ -31,6 +31,7 @@ our ($VERSION, $debug, $lang, $languri);
BEGIN {
$debug = 0;
$VERSION = do { my $REV = (qw$Revision: 121 $)[1]; sprintf("%0.3f", 1 + ($REV/1000)) };
+ our %SERVICE_BLOOM_IGNORE = ('http://dbpedia.org/sparql' => 1); # by default, assume dbpedia doesn't implement k:bloom().
}
######################################################################
@@ -199,6 +200,8 @@ sub execute {
my $handled = 0;
+ our %SERVICE_BLOOM_IGNORE; # keep track of which service calls throw an error so we don't keep trying it...
+
### cooperate with ::Algebra::Service so that if we've already got a stream
### of results from previous patterns, and the next pattern is a remote
### service call, we can try to send along a bloom filter function.
@@ -206,26 +209,27 @@ sub execute {
### function), then fall back on making the call without the filter.
try {
if ($stream and $triple->isa('RDF::Query::Algebra::Service')) {
- my $m = $stream->materialize;
-
- my @vars = $triple->referenced_variables;
- my %svars = map { $_ => 1 } $stream->binding_names;
- my $var = RDF::Query::Node::Variable->new( first { $svars{ $_ } } @vars );
-
- my $f = RDF::Query::Algebra::Service->bloom_filter_for_iterator( $query, $bridge, $bound, $m, $var, 0.001 );
-
- my $new;
- try {
+ unless ($SERVICE_BLOOM_IGNORE{ $triple->endpoint->uri_value }) {
+ # local($RDF::Trine::Iterator::debug) = 1;
+ $stream = $stream->materialize;
+ my $m = $stream;
+
+ my @vars = $triple->referenced_variables;
+ my %svars = map { $_ => 1 } $stream->binding_names;
+ my $var = RDF::Query::Node::Variable->new( first { $svars{ $_ } } @vars );
+
+ my $f = RDF::Query::Algebra::Service->bloom_filter_for_iterator( $query, $bridge, $bound, $m, $var, 0.001 );
+
my $pattern = $triple->add_bloom( $var, $f );
- $new = $pattern->execute( $query, $bridge, $bound, $context, %args );
+ my $new = $pattern->execute( $query, $bridge, $bound, $context, %args );
throw RDF::Query::Error unless ($new);
- } otherwise {
- warn "*** Wasn't able to use :bloom as a FILTER restriction in SERVICE call.\n" if ($debug);
- $new = $triple->execute( $query, $bridge, $bound, $context, %args );
- };
- $stream = RDF::Trine::Iterator::Bindings->join_streams( $m, $new, %args );
- $handled = 1;
+ $stream = $self->join_bnode_streams( $m, $new, $query, $bridge, $bound );
+ $handled = 1;
+ }
}
+ } otherwise {
+ $SERVICE_BLOOM_IGNORE{ $triple->endpoint->uri_value } = 1;
+ warn "*** Wasn't able to use k:bloom as a FILTER restriction in SERVICE call.\n" if ($debug);
};
unless ($handled) {
@@ -245,6 +249,108 @@ sub execute {
return $stream;
}
+=item C<< join_bnode_streams ( $streamA, $streamB, $query, $bridge ) >>
+
+A modified inner-loop join that relies on there being bnode identity hints in
+the data returned by C<< $streamB->extra_result_data >>. These hints are
+combined with locally computed identity values for the items from C<< $streamA >>
+and the streams are merged using a natural join where equality is computed
+either on direct node equality or on any intersection of the identity hints.
+
+The identity hints and locally computed identity values are computed using
+Functional and InverseFunctional property values using N3 syntax. For example,
+a blank node '(r1)' might have identity hints using foaf:mbox_sha1sum such as:
+
+ $extra_result_data = {
+ 'bnode-map' => [ {
+ '(r1)' => ['!<http://xmlns.com/foaf/0.1/mbox_sha1sum>"26fb6400147dcccfda59717ff861db9cb97ac5ec"']
+ } ]
+ };
+
+=cut
+
+sub join_bnode_streams {
+ my $self = shift;
+ my $astream = shift;
+ my $bstream = shift;
+ my $query = shift;
+ my $bridge = shift;
+
+ Carp::confess unless ($astream->isa('RDF::Trine::Iterator::Bindings'));
+ Carp::confess unless ($bstream->isa('RDF::Trine::Iterator::Bindings'));
+
+ ################################################
+ ### BNODE MAP STUFF
+ my $b_extra = $bstream->extra_result_data || {};
+ my (%b_map);
+ foreach my $h (@{ $b_extra->{'bnode-map'} || [] }) {
+ foreach my $id (keys %$h) {
+ my @values = @{ $h->{ $id } };
+ push( @{ $b_map{ $id } }, @values );
+ }
+ }
+ my $b_map = (%b_map) ? \%b_map : undef;
+ ################################################
+
+ my @names = uniq( map { $_->binding_names() } ($astream, $bstream) );
+ my $a = $astream->project( @names );
+ my $b = $bstream->project( @names );
+
+ my @results;
+ my @data = $b->get_all();
+ no warnings 'uninitialized';
+ while (my $rowa = $a->next) {
+ LOOP: foreach my $rowb (@data) {
+ warn "[--JOIN--] " . join(' ', map { my $row = $_; '{' . join(', ', map { join('=', $_, ($row->{$_}) ? $row->{$_}->as_string : '(undef)') } (keys %$row)) . '}' } ($rowa, $rowb)) . "\n" if ($debug);
+ my %keysa = map {$_=>1} (keys %$rowa);
+ my @shared = grep { $keysa{ $_ } } (keys %$rowb);
+ foreach my $key (@shared) {
+ my $val_a = $rowa->{ $key };
+ my $val_b = $rowb->{ $key };
+ my $defined = 0;
+ foreach my $n ($val_a, $val_b) {
+ $defined++ if (defined($n));
+ }
+ if ($defined == 2) {
+ my $equal = $val_a->equal( $val_b );
+ if (not $equal) {
+ my $names = $b_map->{ $val_b->as_string };
+ if ($names) {
+ my $bnames = Set::Scalar->new( @{ $names } );
+ my $anames = Set::Scalar->new( RDF::Query::Algebra::Service->_names_for_node( $val_a, $query, $bridge, {} ) );
+ if ($debug) {
+ warn "anames: $anames\n";
+ warn "bnames: $bnames\n";
+ }
+ if (my $int = $anames->intersection( $bnames )) {
+ warn "node equality based on $int" if ($debug);
+ $equal = 1;
+ }
+ }
+ }
+
+ unless ($equal) {
+ warn "can't join because mismatch of $key (" . join(' <==> ', map {$_->as_string} ($val_a, $val_b)) . ")" if ($debug);
+ next LOOP;
+ }
+ }
+ }
+
+ my $row = { (map { $_ => $rowa->{$_} } grep { defined($rowa->{$_}) } keys %$rowa), (map { $_ => $rowb->{$_} } grep { defined($rowb->{$_}) } keys %$rowb) };
+ if ($debug) {
+ warn "JOINED:\n";
+ foreach my $key (keys %$row) {
+ warn "$key\t=> " . $row->{ $key }->as_string . "\n";
+ }
+ }
+ push(@results, $row);
+ }
+ }
+
+ my $args = $astream->_args;
+ return $astream->_new( \@results, 'bindings', \@names, %$args );
+}
+
1;
__END__
RDF-Query/lib/RDF/Query/Algebra/GroupGraphPattern.pm
@@ -118,7 +118,7 @@ sub add_bloom {
my $pattern = $self->pattern;
my $iri = RDF::Query::Node::Resource->new('http://kasei.us/code/rdf-query/functions/bloom');
- warn "Adding a bloom filter (with " . $bloom->key_count . " items) function to a remote query";
+ warn "Adding a bloom filter (with " . $bloom->key_count . " items) function to a remote query" if ($debug);
my $frozen = $bloom->freeze;
my $literal = RDF::Query::Node::Literal->new( $frozen );
my $expr = RDF::Query::Expression::Function->new( $iri, $var, $literal );
@@ -268,8 +268,14 @@ sub execute {
throw RDF::Query::Error -text => "SERVICE query couldn't get remote content: " . $resp->status_line;
}
my $content = $resp->content;
- warn $content;
- my $stream = RDF::Trine::Iterator->from_string( $content );
+ my $stream = smap {
+ my $bindings = $_;
+ return undef unless ($bindings);
+ my %cast = map {
+ $_ => RDF::Query::Model::RDFTrine::_cast_to_local( $bindings->{ $_ } )
+ } (keys %$bindings);
+ return \%cast;
+ } RDF::Trine::Iterator->from_string( $content );
return $stream;
}
RDF-Query/lib/RDF/Query/Algebra/Service.pm
@@ -16,8 +16,9 @@ use strict;
use warnings;
no warnings 'redefine';
-use Scalar::Util qw(blessed reftype looks_like_number);
+use Scalar::Util qw(blessed reftype refaddr looks_like_number);
+use RDF::Query;
use RDF::Query::Model::RDFTrine;
use RDF::Query::Error qw(:try);
@@ -731,20 +732,18 @@ $RDF::Query::functions{"http://kasei.us/2007/09/functions/warn"} = sub {
my $query = shift;
my $bridge = shift;
my $stream = shift;
- if ($query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' }) {
- foreach my $map (@{ $query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' } }) {
- my ($node, $string) = @$map;
- warn "adding node map: " . $node->as_string . ": " . $string . "\n";
- }
- }
+ warn "bloom filter got result stream\n" if ($debug);
+ my $nodemap = $query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' };
+ $stream->add_extra_result_data('bnode-map', $nodemap);
}
- RDF::Query->add_hook('http://kasei.us/code/rdf-query/hooks/function_init', sub {
+ push( @{ $RDF::Query::hooks{ 'http://kasei.us/code/rdf-query/hooks/function_init' } }, sub {
my $query = shift;
my $function = shift;
- if ($function eq $BLOOM_URL) {
+ if ($function->uri_value eq $BLOOM_URL) {
+ $query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' } ||= {};
$query->add_hook_once( 'http://kasei.us/code/rdf-query/hooks/post-execute', \&_BLOOM_ADD_NODE_MAP_TO_STREAM, "${BLOOM_URL}#add_node_map" );
}
- });
+ } );
$RDF::Query::functions{"http://kasei.us/code/rdf-query/functions/bloom"} = sub {
my $query = shift;
my $bridge = shift;
@@ -767,7 +766,8 @@ $RDF::Query::functions{"http://kasei.us/2007/09/functions/warn"} = sub {
my $ok = $bloom->check( $string );
warn "-> ok\n" if ($ok and $debug);
if ($ok) {
- push( @{ $query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' } }, [ $value, $string ] );
+ my $nodemap = $query->{_query_cache}{ $BLOOM_URL }{ 'nodemap' };
+ push( @{ $nodemap->{ $value->as_string } }, $string );
return RDF::Query::Node::Literal->new('true', undef, 'http://www.w3.org/2001/XMLSchema#boolean');
}
}
RDF-Query/lib/RDF/Query/Functions.pm
@@ -6,50 +6,93 @@ use Test::More;
use lib qw(. t);
BEGIN { require "models.pl"; }
-my @files;
-my @models = test_models( @files );
-
+my $tests = 25;
eval { require LWP::Simple };
if ($@) {
plan skip_all => "LWP::Simple is not available for loading <http://...> URLs";
return;
} elsif (not exists $ENV{RDFQUERY_NO_NETWORK}) {
- plan tests => 14;
+ plan tests => $tests;
+} elsif (not exists $ENV{RDFQUERY_DEV_TESTS}) {
+ plan skip_all => 'Developer tests. Set RDFQUERY_DEV_TESTS to run these tests.';
+ return;
} else {
plan skip_all => 'No network. Unset RDFQUERY_NO_NETWORK to run these tests.';
return;
}
-my $loaded = use_ok( 'RDF::Query' );
-BAIL_OUT( 'RDF::Query not loaded' ) unless ($loaded);
+use RDF::Query;
-SKIP: {
- unless ($ENV{RDFQUERY_DEV_TESTS}) {
- skip "developer network tests", 4;
- }
+{
+ print "# join using default graph (local rdf) and remote SERVICE (kasei.us), joining on bnode\n";
+ my $file = URI::file->new_abs( 'data/bnode-person.rdf' );
my $query = RDF::Query->new( <<"END", undef, undef, 'sparqlp' );
PREFIX foaf: <http://xmlns.com/foaf/0.1/>
- SELECT DISTINCT *
- FROM <http://kasei.us/about/foaf.xrdf>
+ PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+ SELECT DISTINCT ?name ?nick
+ FROM <$file>
WHERE {
?p a foaf:Person ; foaf:name ?name .
- FILTER <http://kasei.us/code/rdf-query/functions/bloom>( ?p, "BAcEMTIzNAQEBAgRDUJsb29tOjpGaWx0ZXIDCAAAAAoEAAAAAAgAAABibGFua3ZlYweamZmZmZmpPwoAAABlcnJvcl9yYXRlCggAAAFAAAgCAAYAAABmaWx0ZXIIhA4AAABudW1faGFzaF9mdW5jcwiBCQAAAGtleV9jb3VudAi/DQAAAGZpbHRlcl9sZW5ndGgIiggAAABjYXBhY2l0eQQCBAAAAAoRMC4zNzkxOTU0NDEyOTczOTQKETAuMjQ0MjQ5NjM2ODQ2NjkzChEwLjg1MTQ5ODA5NTE3NjMxMQoSMC4wNzgyMTI0NDE5NjY1MjA5BQAAAHNhbHRz" ) .
+ SERVICE <http://kasei.us/sparql> {
+ ?p foaf:nick ?nick
+ }
}
END
my $stream = $query->execute();
isa_ok( $stream, 'RDF::Trine::Iterator' );
- my $d = $stream->next;
- isa_ok( $d, 'HASH' );
- isa_ok( $d->{ name }, 'RDF::Trine::Node::Literal' );
- is( $d->{name}->literal_value, 'Gregory Todd Williams', 'expected value passed through bloom filter' );
+ my $count = 0;
+ while (my $d = $stream->next) {
+ isa_ok( $d->{nick}, 'RDF::Query::Node::Literal' );
+ like( $d->{name}->literal_value, qr/^(Adam Pisoni|Gregory Todd Williams)$/, 'got name from local file (joined on a bnode)' );
+ like( $d->{nick}->literal_value, qr/^(wonko)$/, 'got nick from SERVICE (joined on a bnode)' );
+ $count++;
+ }
+ is( $count, 1, 'expected result count' );
}
-
-SKIP: {
- unless ($ENV{RDFQUERY_DEV_TESTS}) {
- skip "developer network tests", 3;
+{
+ my $file = URI::file->new_abs( 'data/bnode-person.rdf' );
+
+ ### This filter contains greg and adam, identified by a primaryTopic page and an email sha1sum, respectively:
+ ### !<http://xmlns.com/foaf/0.1/mbox_sha1sum>"26fb6400147dcccfda59717ff861db9cb97ac5ec"
+ ### ^<http://xmlns.com/foaf/0.1/primaryTopic><http://kasei.us/>
+ my $sparql = <<"END";
+ PREFIX foaf: <http://xmlns.com/foaf/0.1/>
+ PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+ PREFIX k: <http://kasei.us/code/rdf-query/functions/>
+ SELECT DISTINCT *
+ FROM <$file>
+ WHERE {
+ ?p a foaf:Person ; foaf:name ?name .
+ FILTER k:bloom( ?p, "AAAAAgAAAB0AAAACAAAACgAAAAQAAAAFZJREZxMt9AekxG8jP9HulJISKVbFzR2rP9boeVledzCy\\nIqE/jWf8NvjRYRUwLjAwMQ==\\n" ) .
+ }
+END
+ {
+ print "# bgp using default graph (local rdf) with k:bloom FILTER produces bnode identity hints in XML results\n";
+ my $query = RDF::Query->new( $sparql, undef, undef, 'sparqlp' );
+ my $stream = $query->execute();
+ my $xml = $stream->as_xml;
+ like( $xml, qr#<extrakey id="[^"]+">!<http://xmlns.com/foaf/0.1/mbox_sha1sum>"26fb6400147dcccfda59717ff861db9cb97ac5ec"</extrakey>#sm, 'xml serialization has good looking bnode map' );
}
+ {
+ print "# bgp using default graph (local rdf) with k:bloom FILTER\n";
+ my $query = RDF::Query->new( $sparql, undef, undef, 'sparqlp' );
+ my $stream = $query->execute();
+ isa_ok( $stream, 'RDF::Trine::Iterator' );
+
+ my $count = 0;
+ while (my $d = $stream->next) {
+ isa_ok( $d->{name}, 'RDF::Query::Node::Literal' );
+ like( $d->{name}->literal_value, qr/^(Adam Pisoni|Gregory Todd Williams)$/, 'expected name passed from person passing through bloom filter' );
+ $count++;
+ }
+ is( $count, 2, 'expected result count' );
+ }
+}
+
+{
+ print "# join using default graph (remote rdf) and remote SERVICE (kasei.us), joining on IRI\n";
my $query = RDF::Query->new( <<"END", undef, undef, 'sparqlp' );
PREFIX foaf: <http://xmlns.com/foaf/0.1/>
SELECT DISTINCT *
@@ -68,22 +111,14 @@ END
isa_ok( $stream, 'RDF::Trine::Iterator' );
my $d = $stream->next;
isa_ok( $d, 'HASH' );
- is_deeply( $d, {
- 'p' => bless( [
- 'URI',
- 'http://kasei.us/about/foaf.xrdf#greg',
- ], 'RDF::Trine::Node::Resource' ),
- 'name' => bless( [
- 'LITERAL',
- 'Gregory Todd Williams'
- ], 'RDF::Trine::Node::Literal' )
- } );
+ isa_ok( $d->{p}, 'RDF::Trine::Node::Resource' );
+ is( $d->{p}->uri_value, 'http://kasei.us/about/foaf.xrdf#greg', 'expected person uri' );
+ isa_ok( $d->{name}, 'RDF::Trine::Node::Literal' );
+ is( $d->{name}->literal_value, 'Gregory Todd Williams', 'expected person name' );
}
-SKIP: {
- unless ($ENV{RDFQUERY_DEV_TESTS}) {
- skip "developer network tests", 3;
- }
+{
+ print "# join using default graph (remote rdf) and remote SERVICE (dbpedia), joining on IRI\n";
my $query = RDF::Query->new( <<"END", undef, undef, 'sparqlp' );
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
SELECT DISTINCT *
@@ -95,48 +130,17 @@ SKIP: {
}
SERVICE <http://dbpedia.org/sparql> {
?thing a <http://dbpedia.org/class/yago/Island109316454>
+ FILTER( REGEX( STR(?thing), "http://dbpedia.org/resource/V" ) ) .
}
}
END
my $stream = $query->execute();
isa_ok( $stream, 'RDF::Trine::Iterator' );
my $d = $stream->next;
-
- isa_ok( $d, 'HASH' );
- is_deeply( $d, {
- 'label' => bless( [
- 'LITERAL',
- 'Vancouver Island',
- 'en',
- undef
- ], 'RDF::Trine::Node::Literal' ),
- 'thing' => bless( [
- 'URI',
- 'http://dbpedia.org/resource/Vancouver_Island'
- ], 'RDF::Trine::Node::Resource' )
- } );
-}
-
-
-SKIP: {
- unless ($ENV{RDFQUERY_DEV_TESTS}) {
- skip "developer network tests", 3;
- }
- my $query = RDF::Query->new( <<"END", undef, undef, 'sparqlp' );
-PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
-SELECT *
-WHERE {
- SERVICE <http://dbpedia.org/sparql> {
- ?thing a <http://dbpedia.org/class/yago/Island109316454>
- }
-}
-END
- my $stream = $query->execute();
- isa_ok( $stream, 'RDF::Trine::Iterator' );
- my $d = $stream->next;
isa_ok( $d, 'HASH' );
-
- my @values = values %$d;
- isa_ok( $values[0], 'RDF::Trine::Node' );
+ isa_ok( $d->{label}, 'RDF::Trine::Node::Literal' );
+ is( $d->{label}->literal_value, 'Vancouver Island', 'expected (island) name' );
+ is( $d->{label}->literal_value_language, 'en', 'expected (island) name language' );
+ isa_ok( $d->{thing}, 'RDF::Trine::Node::Resource' );
+ is( $d->{thing}->uri_value, 'http://dbpedia.org/resource/Vancouver_Island', 'expected (island) uri' );
}
-
RDF-Query/t/31-service.t
@@ -39,7 +39,7 @@ use XML::Twig;
use Data::Dumper;
use Carp qw(carp);
use List::MoreUtils qw(uniq);
-use Scalar::Util qw(blessed reftype);
+use Scalar::Util qw(blessed reftype refaddr);
use RDF::Trine::Node;
@@ -183,6 +183,7 @@ sub from_string {
my $value;
my @results;
my $result = {};
+ my (%extrakeys, %extra);
my $twig = XML::Twig->new(
twig_handlers => {
@@ -197,6 +198,8 @@ sub from_string {
$value = RDF::Trine::Node::Literal->new( $_->text, $lang, $dt )
},
boolean => sub { $bool = ($_->text eq 'true') ? 1 : 0 },
+ extra => sub { push( @{ $extra{ $_->att('name') } }, { %extrakeys } ); %extrakeys = (); },
+ extrakey => sub { push(@{ $extrakeys{ $_->att('id') } }, $_->text); },
},
);
$twig->parse( $string );
@@ -204,7 +207,13 @@ sub from_string {
if (defined($bool)) {
return RDF::Trine::Iterator::Boolean->new( [$bool] );
} else {
- return RDF::Trine::Iterator::Bindings->new( \@results, \@vars );
+ my $bindings = RDF::Trine::Iterator::Bindings->new( \@results, \@vars, extra_result_data => \%extra );
+# foreach my $tag (keys %extra) {
+# foreach my $value (@{ $extra{ $tag } }) {
+# $bindings->add_extra_result_data( $tag, $value );
+# }
+# }
+ return $bindings;
}
}
@@ -216,8 +225,8 @@ Returns the next item in the stream.
=cut
-sub next { $_[0]->next_result }
-sub next_result {
+sub next_result { $_[0]->next }
+sub next {
my $self = shift;
return if ($self->{_finished});
@@ -227,14 +236,6 @@ sub next_result {
$self->{_finished} = 1;
}
- my $args = $self->_args;
-# if ($args->{named}) {
-# if ($self->_bridge->supports('named_graph')) {
-# my $bridge = $self->_bridge;
-# $args->{context} = $bridge->get_context( $self->{_stream}, %$args );
-# }
-# }
-
$self->{_open} = 1;
$self->{_row} = $value;
return $value;
@@ -295,22 +296,6 @@ sub close {
return;
}
-# =item C<< context >>
-#
-# Returns the context node of the current result (if applicable).
-#
-# =cut
-#
-# sub context {
-# my $self = shift;
-# my $args = $self->_args;
-# my $bridge = $args->{bridge};
-# my $stream = $self->{_stream};
-# my $context = $bridge->get_context( $stream, %$args );
-# return $context;
-# }
-
-
=item C<< concat ( $stream ) >>
Returns a new stream resulting from the concatenation of the referant and the
@@ -388,6 +373,27 @@ sub join_streams {
Carp::confess unless ($astream->isa('RDF::Trine::Iterator::Bindings'));
Carp::confess unless ($bstream->isa('RDF::Trine::Iterator::Bindings'));
+ ################################################
+ ### BNODE MAP STUFF
+ my $a_extra = $astream->extra_result_data || {};
+ my $b_extra = $bstream->extra_result_data || {};
+ my (%a_map, %b_map);
+ foreach my $h (@{ $a_extra->{'bnode-map'} || [] }) {
+ foreach my $id (keys %$h) {
+ my @values = @{ $h->{ $id } };
+ push( @{ $a_map{ $id } }, @values );
+ }
+ }
+ foreach my $h (@{ $b_extra->{'bnode-map'} || [] }) {
+ foreach my $id (keys %$h) {
+ my @values = @{ $h->{ $id } };
+ push( @{ $b_map{ $id } }, @values );
+ }
+ }
+ my $a_map = (%a_map) ? \%a_map : undef;
+ my $b_map = (%b_map) ? \%b_map : undef;
+ ################################################
+
my @names = uniq( map { $_->binding_names() } ($astream, $bstream) );
my $a = $astream->project( @names );
my $b = $bstream->project( @names );
@@ -397,7 +403,7 @@ sub join_streams {
no warnings 'uninitialized';
while (my $rowa = $a->next) {
LOOP: foreach my $rowb (@data) {
- warn "[--JOIN--] " . join(' ', map { my $row = $_; '{' . join(', ', map { join('=',$_,$row->{$_}->as_string) } (keys %$row)) . '}' } ($rowa, $rowb)) . "\n" if ($debug);
+ warn "[--JOIN--] " . join(' ', map { my $row = $_; '{' . join(', ', map { join('=', $_, ($row->{$_}) ? $row->{$_}->as_string : '(undef)') } (keys %$row)) . '}' } ($rowa, $rowb)) . "\n" if ($debug);
my %keysa = map {$_=>1} (keys %$rowa);
my @shared = grep { $keysa{ $_ } } (keys %$rowb);
foreach my $key (@shared) {
@@ -408,7 +414,35 @@ sub join_streams {
$defined++ if (defined($n));
}
if ($defined == 2) {
- unless ($val_a->equal($val_b)) {
+ my $equal = $val_a->equal( $val_b );
+ if (not $equal) {
+ my $query = $args{ query };
+ my $bridge = $args{ bridge };
+ if ($query and $bridge) {
+ warn 'join values and bnode maps: ' . Dumper($val_a, $val_b, $a_map, $b_map) if ($a_map or $b_map);
+ if ($a_map and $val_a->isa('RDF::Trine::Node::Blank')) {
+ my $anames = Set::Scalar->new( @{ $a_map{ $val_a->blank_identifier } } );
+ my $bnames = Set::Scalar->new( RDF::Query::Algebra::Service->_names_for_node( $val_b, $query, $bridge, {} ) );
+ if (my $int = $anames->intersection( $bnames )) {
+ warn "node equality based on $int";
+ $equal = 1;
+ }
+ } elsif ($b_map and $val_b->isa('RDF::Trine::Node::Blank')) {
+ my $bnames = Set::Scalar->new( @{ $b_map{ $val_b->blank_identifier } } );
+ my $anames = Set::Scalar->new( RDF::Query::Algebra::Service->_names_for_node( $val_a, $query, $bridge, {} ) );
+ warn "anames: $anames\n";
+ warn "bnames: $bnames\n";
+ if (my $int = $anames->intersection( $bnames )) {
+ warn "node equality based on $int";
+ $equal = 1;
+ }
+ }
+ } else {
+ Carp::cluck "no query,bridge in args" if ($debug);
+ }
+ }
+
+ unless ($equal) {
warn "can't join because mismatch of $key (" . join(' <==> ', map {$_->as_string} ($val_a, $val_b)) . ")" if ($debug);
next LOOP;
}
@@ -491,7 +525,7 @@ sub construct_args {
my $self = shift;
my $type = $self->type;
my $args = $self->_args || {};
- return ($type, []);
+ return ($type, [], %$args);
}
=begin private
@@ -550,6 +584,24 @@ sub _stream {
}
+=item C<< add_extra_result_data ( $tag, \%data ) >>
+
+=cut
+
+sub add_extra_result_data {
+ my $self = shift;
+ my $tag = shift;
+ my $data = shift;
+ push( @{ $self->_args->{ extra_result_data }{ $tag } }, $data );
+}
+
+sub extra_result_data {
+ my $self = shift;
+ my $args = $self->_args;
+ my $extra = $args->{ extra_result_data };
+ return $extra;
+}
+
RDF-Trine/lib/RDF/Trine/Iterator.pm
@@ -39,9 +39,8 @@ use RDF::Trine::Iterator qw(smap);
use base qw(RDF::Trine::Iterator);
our ($REVISION, $VERSION, $debug);
-use constant DEBUG => 0;
BEGIN {
- $debug = DEBUG;
+ $debug = 0;
$REVISION = do { my $REV = (qw$Revision: 293 $)[1]; sprintf("%0.3f", 1 + ($REV/1000)) };
$VERSION = '1.000';
}
@@ -149,7 +148,7 @@ sub join_streams {
my $a_sort = join(',', $a->sorted_by);
my $b_sort = join(',', $b->sorted_by);
- if (DEBUG) {
+ if ($debug) {
warn '---------------------------';
warn 'REQUESTED SORT in JOIN: ' . Dumper($req_sort);
warn 'JOIN STREAM SORTED BY: ' . Dumper($a_sort);
@@ -157,17 +156,17 @@ sub join_streams {
}
my $actual_sort;
if (substr( $a_sort, 0, length($req_sort) ) eq $req_sort) {
- warn "first stream is already sorted. using it in the outer loop.\n" if (DEBUG);
+ warn "first stream is already sorted. using it in the outer loop.\n" if ($debug);
} elsif (substr( $b_sort, 0, length($req_sort) ) eq $req_sort) {
- warn "second stream is already sorted. using it in the outer loop.\n" if (DEBUG);
+ warn "second stream is already sorted. using it in the outer loop.\n" if ($debug);
($a,$b) = ($b,$a);
} else {
my $a_common = join('!', $a_sort, $req_sort);
my $b_common = join('!', $b_sort, $req_sort);
if ($a_common =~ qr[^([^!]+)[^!]*!\1]) { # shared prefix between $a_sort and $req_sort?
- warn "first stream is closely sorted ($1).\n" if (DEBUG);
+ warn "first stream is closely sorted ($1).\n" if ($debug);
} elsif ($b_common =~ qr[^([^!]+)[^!]*!\1]) { # shared prefix between $b_sort and $req_sort?
- warn "second stream is closely sorted ($1).\n" if (DEBUG);
+ warn "second stream is closely sorted ($1).\n" if ($debug);
($a,$b) = ($b,$a);
}
}
@@ -175,7 +174,7 @@ sub join_streams {
}
my $stream = $self->SUPER::join_streams( $a, $b, %args );
- warn "JOINED stream is sorted by: " . join(',', @join_sorted_by) . "\n" if (DEBUG);
+ warn "JOINED stream is sorted by: " . join(',', @join_sorted_by) . "\n" if ($debug);
$stream->{sorted_by} = \@join_sorted_by;
return $stream;
}
@@ -386,10 +385,29 @@ END
last if ($max_result_size and ++$count >= $max_result_size);
} continue { $self->next_result }
- $xml .= <<"EOT";
-</results>
-</sparql>
-EOT
+ $xml .= "</results>\n";
+
+ if (my $extra = $self->extra_result_data) {
+ foreach my $tag (keys %$extra) {
+ $xml .= qq[<extra name="${tag}">\n];
+ my $value = $extra->{ $tag };
+ foreach my $e (@$value) {
+ foreach my $k (keys %$e) {
+ my $v = $e->{ $k };
+ my @values = @$v;
+ foreach ($k, @values) {
+ s/&/&/g;
+ s/</</g;
+ s/"/"/g;
+ }
+ $xml .= qq[\t<extrakey id="$k">] . join(',', @values) . qq[</extrakey>\n];
+ }
+ }
+ $xml .= "</extra>\n";
+ }
+ }
+
+ $xml .= "</sparql>\n";
return $xml;
}
RDF-Trine/lib/RDF/Trine/Iterator/Bindings.pm
7aaedae28f61e48f4eed20239c35ae39dfc70481
ba42ed1f3e09aecf45dfc58827cba66f9fb76d88
Gregory Williams
samofool@tiu.local
http://github.com/kasei/perlrdf/commit/e27a1d7a520ff42a5d079ed4cd6267e112f353cc
e27a1d7a520ff42a5d079ed4cd6267e112f353cc
2008-03-10T19:12:38-07:00
2008-03-10T19:12:38-07:00
Merge branch 'service-bnode-join'
6de85651d51d01688590513da4470508dce1cae1
Gregory Williams
samofool@tiu.local