Database-Join
view release on metacpan or search on metacpan
lib/Database/Join.pm view on Meta::CPAN
=head3 MESSAGES
C<updated()> does not emit any warnings or errors. Component databases that do
not implement C<updated()>, or whose C<updated()> throws, are silently skipped;
only defined return values contribute to the maximum. If no component database
implements C<updated()>, C<undef> is returned (same as C<List::Util::max> on an
empty list).
=cut
sub updated {
my ($self) = @_;
my @timestamps;
for my $db (@{ $self->{_dbs} }) {
my $ts;
do { local $@; $ts = eval { $db->updated() } };
push @timestamps, $ts if defined $ts;
}
return max(@timestamps);
}
t/aggregate.t view on Meta::CPAN
my ($class, %args) = @_;
return bless {
_cols => $args{cols} // [],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
id => $args{id} // $args{cols}[0],
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return undef }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$criteria //= {};
my @result;
ROW: for my $row (@{ $self->{_rows} }) {
for my $col (keys %{$criteria}) {
my $want = $criteria->{$col};
my $got = $row->{$col};
if (ref($want) eq 'HASH') {
for my $op (keys %{$want}) {
t/collision-prefix.t view on Meta::CPAN
return bless {
cols => $args{cols} // ['entry'],
rows => $args{rows} // [],
schema => $args{schema} // {},
updated => $args{updated} // 1,
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return $_[0]->{schema} }
sub updated { return $_[0]->{updated} }
sub set_logger { return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$criteria //= {};
my @out;
ROW: for my $row (@{ $self->{rows} }) {
for my $col (keys %{$criteria}) {
my $val = $criteria->{$col};
if (ref $val eq 'HASH') {
t/data-flow.t view on Meta::CPAN
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1_000_000,
_received => [], # [ { hashref => ..., refaddr => ... }, ... ]
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
push @{ $self->{_received} }, {
hashref => $criteria,
refaddr => Scalar::Util::refaddr($criteria),
};
# Simple equality filter (no operator hashrefs needed for DU tests)
my @rows = @{ $self->{_rows} };
t/data-flow.t view on Meta::CPAN
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1_000_000,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{logger} = $_[1]; return $_[0] }
sub selectall_arrayref { return $_[0]->{_rows} } # always returns all rows
sub DESTROY {}
}
# ---------------------------------------------------------------------------
# DFCountingDA: records how many times selectall_arrayref is called.
# Applies simple equality filtering (same as DFRecordingDA) so filter criteria
# passed at spill time actually restrict the spilled rows.
# count() is defined directly in this package (not inherited) so that the
t/data-flow.t view on Meta::CPAN
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1_000_000,
_calls => 0,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{logger} = $_[1]; return $_[0] }
sub call_count { return $_[0]->{_calls} }
sub count { return scalar @{ $_[0]->{_rows} } } # direct def for auto-threshold
sub selectall_arrayref {
my ($self, $criteria) = @_;
$self->{_calls}++;
my @rows = @{ $self->{_rows} };
for my $col (keys %{ $criteria // {} }) {
my $v = $criteria->{$col};
t/data-flow.t view on Meta::CPAN
# ---------------------------------------------------------------------------
# DFNoTsDA: DA whose updated() returns undef (not a die/croak).
# _build_sqlite_cache stores undef for the timestamp.
# _cache_fresh: `$cached_ts = undef // next` skips the staleness check,
# keeping the cache valid indefinitely for this source.
# ---------------------------------------------------------------------------
{
package DFNoTsDA;
use parent -norequire, 'DFMinimalDA';
sub updated { return undef }
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
sub _two_db_join {
my (%opts) = @_;
my $prim = DFMinimalDA->new(
cols => ['entry', 'name'],
rows => [
t/data-flow.t view on Meta::CPAN
# Without join_map or filters, AUTOLOAD must still resolve $db and delegate.
# Since DFMinimalDA does not implement AUTOLOAD itself, any column shortcut
# call would fail if the direct-delegation path were broken.
# We verify by using a custom DA that DOES define the column as a method.
{
package DirectDA;
use parent -norequire, 'Database::Abstraction';
sub new { bless { id=>'entry', _cols=>['entry','tag'], _rows=>[], _schema=>{}, _ts=>1 }, shift }
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return 1 }
sub set_logger { $_[0]->{logger} = $_[1]; return $_[0] }
sub selectall_arrayref { return $_[0]->{_rows} }
sub tag { return 'direct-value' } # method that AUTOLOAD will delegate to
sub DESTROY {}
}
my $da = DirectDA->new();
my $j = Database::Join->new(databases => [$da], join_column => 'entry');
# No join_map or filters: AUTOLOAD takes the direct path and calls $da->tag()
my $val = $j->tag();
is $val, 'direct-value',
t/data-flow.t view on Meta::CPAN
};
# ---------------------------------------------------------------------------
# DFThrowUpdDA: DA whose updated() always throws.
# Used to verify that _joined_query / updated() resilience silently skips it.
# ---------------------------------------------------------------------------
{
package DFThrowUpdDA;
use parent -norequire, 'DFMinimalDA';
use Carp qw(croak);
sub updated { croak 'simulated updated() failure' }
sub DESTROY {}
}
# ---------------------------------------------------------------------------
# DFBareDA: duck-type DA with no Database::Abstraction parent and no updated()
# method. Proves that code paths that check `$db->can('updated')` (or use
# eval-guarded calls) handle a missing method without croaking.
# ---------------------------------------------------------------------------
{
package DFBareDA;
t/distinct-mandatory.t view on Meta::CPAN
# ---------------------------------------------------------------------------
package MinimalDA;
use parent -norequire, 'Database::Abstraction';
sub new {
my ($class, %args) = @_;
return bless { _cols => $args{cols}, _rows => $args{rows} }, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return undef }
sub set_logger { return $_[0] }
sub selectall_arrayref {
my ($self, $crit) = @_;
$crit //= {};
my @rows = @{ $self->{_rows} };
for my $col (keys %{$crit}) {
my $val = $crit->{$col};
if (!ref $val) {
@rows = grep { defined $_->{$col} && $_->{$col} eq $val } @rows;
} elsif (ref $val eq 'HASH') {
cols => $args{cols} // ['entry'],
rows => $args{rows} // [],
id => $args{id} // 'entry',
updated => $args{updated} // $UPDATED,
schema => $args{schema} // {},
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return $_[0]->{schema} }
sub updated { return $_[0]->{updated} }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$criteria //= {};
my @results;
ROW: for my $row (@{ $self->{rows} }) {
for my $col (keys %{$criteria}) {
my $val = $criteria->{$col};
if (ref $val eq 'HASH') {
for my $op (keys %{$val}) {
t/edge_cases.t view on Meta::CPAN
_schema => $args{schema}, # deliberately no default -- undef is valid input
_ts => $args{updated} // 1_000_000,
_return => $args{return},
_do_ret => (exists $args{return}), # distinguish return=>undef from absent key
_croak => $args{croak_msg},
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
croak($self->{_croak}) if $self->{_croak};
return $self->{_return} if $self->{_do_ret};
my @rows = @{ $self->{_rows} };
for my $col (keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
if (ref($val) eq 'HASH') {
t/edge_cases.t view on Meta::CPAN
my ($class, %args) = @_;
return bless {
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return 1 }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
# HOSTILE: overwrite the '>' operator inside the entry criteria hashref to
# attempt to corrupt sibling databases' criteria for the same broadcast key.
if (ref($criteria) eq 'HASH' && ref($criteria->{entry}) eq 'HASH') {
$criteria->{entry}{'>'} = 999_999;
}
return $self->{_rows};
t/edge_cases.t view on Meta::CPAN
return bless {
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_ts => $args{updated} // 1_000_000,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub count { return scalar @{ $_[0]->{_rows} } }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
for my $col (keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
next if ref $val;
@rows = grep { defined $_->{$col} && $_->{$col} eq $val } @rows;
}
t/edge_cases.t view on Meta::CPAN
sub new {
my ($class, %args) = @_;
return bless {
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { croak 'schema() simulated failure' }
sub updated { return 1 }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref { return $_[0]->{_rows} }
sub DESTROY {}
}
# BareEdgeDA: minimal duck-type DA with no updated() method at all.
# Proves updated() resilience when a component has no timestamp support.
{
package BareEdgeDA;
sub new {
t/edge_cases.t view on Meta::CPAN
sub selectall_arrayref { return $_[0]->{_rows} }
sub DESTROY {}
# No updated() method -- deliberately absent
}
# ThrowUpdEdgeDA: DA whose updated() throws, to probe updated() resilience.
{
package ThrowUpdEdgeDA;
use parent -norequire, 'MockEdgeDA';
use Carp qw(croak);
sub updated { croak 'simulated updated() failure' }
sub DESTROY {}
}
# ===========================================================================
# S17: limit / offset -- hostile inputs
# ===========================================================================
# 0.007.0: limit => N (positive integer), offset => M (non-negative integer).
# Invalid values emit carp and are treated as absent (no crash, no die).
subtest 'limit/offset: negative limit carps and is treated as absent' => sub {
t/function.t view on Meta::CPAN
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
# Simple equality filter for tests that need it
for my $col (keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
next if ref $val;
@rows = grep { defined $_->{$col} && $_->{$col} eq $val } @rows;
}
t/function.t view on Meta::CPAN
'_cache_fresh returns 0 after source updated() timestamp changes');
};
subtest '_cache_fresh: returns 1 when source has no updated() (timestamp skipped)' => sub {
plan tests => 1;
# A source DA without an updated() method should not invalidate the cache.
# _cache_fresh skips the timestamp check for it, leaving the cache valid.
{
package MinimalDA::NoTS;
use parent -norequire, 'MinimalDA';
sub updated { die 'no updated()' } # simulates a DA that croaks
}
my $db = bless MinimalDA->new(cols => [$JC], rows => []), 'MinimalDA::NoTS';
my $wb = Database::Join::WhiteBox->new(
databases => [$db],
join_column => $JC,
backend => 'sqlite',
);
$wb->expose_build_sqlite_cache();
# _sqlite_cache{updated}{0} is undef because updated() died → next → skip.
ok($wb->expose_cache_fresh(),
t/integration.t view on Meta::CPAN
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1_000_000,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
for my $col (keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
if (ref($val) eq 'HASH') {
for my $op (keys %{$val}) {
my $v = $val->{$op};
t/integration.t view on Meta::CPAN
sub schema { return {} }
sub selectall_arrayref { return $_[0]->{_rows} }
sub DESTROY {}
}
# IntThrowUpdDA: inherits from InMemDA but overrides updated() to die,
# exercising the "skip DAs whose updated() throws" branch.
{
package IntThrowUpdDA;
our @ISA = ('InMemDA');
sub updated { die "simulated updated() failure\n" }
sub DESTROY {}
}
subtest 'updated(): returns undef when no component DA implements updated()' => sub {
plan tests => 1;
# IntBareDA has no updated() method; the join must return undef, not croak.
my $db = IntBareDA->new(
cols => [$JC, 'x'],
rows => [ { entry => 'K', x => 1 } ],
);
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1_000_000,
_calls => 0,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub call_count { return $_[0]->{_calls} }
sub set_logger { $_[0]->{logger} = $_[1]; return $_[0] }
# Defined directly so defined(&{"LogicDA::count"}) is TRUE -- needed for
# backend='auto' threshold probe which distinguishes own-defined vs inherited.
sub count { return scalar @{ $_[0]->{_rows} } }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$self->{_calls}++;
return \@rows;
}
sub DESTROY {}
}
# DA that returns undef from updated() -- used for cache-skip tests.
{
package LogicNoTsDA;
use parent -norequire, 'LogicDA';
sub updated { return undef }
}
# Minimal stub that has selectall_arrayref but NOT columns().
# Used by L14 to prove that the P1 invariant is enforced at every entry point.
{
package NoCols;
sub new { bless {}, shift }
sub selectall_arrayref { return [] }
sub DESTROY {}
}
sub DESTROY {}
# No updated() -- deliberately absent
}
# LogicThrowUpdDA: DA whose updated() throws unconditionally.
# Used by L22 to prove that throwing DAs are skipped and do not propagate errors.
{
package LogicThrowUpdDA;
use parent -norequire, 'LogicDA';
use Carp qw(croak);
sub updated { croak 'simulated updated() failure' }
sub DESTROY {}
}
# ===========================================================================
# Shared fixtures: primary has keys {A,B,C}; secondary has keys {B,D}.
# ===========================================================================
Readonly::Hash my %ROW => (
pA => { entry => 'A', name => 'alice' },
pB => { entry => 'B', name => 'bob' },
cols => $a{cols} // ['entry'],
rows => $a{rows} // [],
id => $a{id}, # may be undef to test fallback path
schema => $a{schema} // {},
updated => $a{updated} // 1,
_logger => undef,
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return $_[0]->{schema} } # may return undef explicitly
sub updated { return $_[0]->{updated} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub get_logger { return $_[0]->{_logger} }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$criteria //= {};
my @out;
ROW: for my $row (@{ $self->{rows} }) {
for my $col (keys %{$criteria}) {
my $val = $criteria->{$col};
return bless {
cols => $a{cols} // ['entry', 'score'],
rows => $a{rows} // [],
id => $a{id} // 'entry',
_calls => 0,
_logger => undef,
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return {} }
sub updated { return 1 }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub get_logger { return $_[0]->{_logger} }
sub calls { return $_[0]->{_calls} }
# Real column method — AUTOLOAD delegates here on the direct path
sub score {
my ($self, @args) = @_;
$self->{_calls}++;
return wantarray ? (90, 70) : 90;
}
sub new {
my ($class, %args) = @_;
return bless {
_cols => $args{cols} // [],
_rows => $args{rows} // [],
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return 1 }
sub set_logger { return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
for my $col (CORE::keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
next if ref $val;
@rows = grep { defined $_->{$col} && $_->{$col} eq $val } @rows;
}
t/redis_live.t view on Meta::CPAN
sub new {
my ($class, %args) = @_;
return bless {
_cols => $args{cols} // [],
_rows => $args{rows} // [],
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return {} }
sub updated { return 1 }
sub set_logger { return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
for my $col (CORE::keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
next if ref $val;
@rows = grep { defined $_->{$col} && $_->{$col} eq $val } @rows;
}
t/sqlite-backend.t view on Meta::CPAN
{
package UpdatableDA;
use parent -norequire, 'CountingDA';
sub new {
my ($class, %args) = @_;
my $self = $class->SUPER::new(%args);
$self->{_ts} = $args{ts} // 1000;
return $self;
}
sub updated { return $_[0]->{_ts} }
sub set_updated { $_[0]->{_ts} = $_[1] }
sub set_rows { $_[0]->{_rows} = $_[1] }
1;
}
subtest 'cache invalidation: cache rebuilt when source updated() timestamp changes' => sub {
my $da_a = UpdatableDA->new(
cols => [qw(id name)],
rows => [{ id => 'k1', name => 'Alice' }],
ts => 1000,
t/transaction.t view on Meta::CPAN
schema => $args{schema} // {},
updated => $args{updated} // 1,
_fail => $args{fail} // 0,
_calls => 0,
_logger => undef,
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return $_[0]->{schema} }
sub updated { return $_[0]->{updated} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub get_logger { return $_[0]->{_logger} }
sub call_count { return $_[0]->{_calls} }
sub set_fail { $_[0]->{_fail} = $_[1]; return $_[0] }
sub reset_calls { $_[0]->{_calls} = 0; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
$self->{_calls}++;
croak 'TransactionDA: simulated mid-flight failure' if $self->{_fail};
t/transition.t view on Meta::CPAN
rows => $a{rows} // [],
id => $a{id} // 'entry',
schema => $a{schema} // {},
updated => $a{updated} // 1,
_calls => 0,
_logger => undef,
}, $class;
}
sub columns { return $_[0]->{cols} }
sub schema { return $_[0]->{schema} // {} }
sub updated { return $_[0]->{updated} }
sub call_count { return $_[0]->{_calls} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $crit) = @_;
$self->{_calls}++;
$crit //= {};
my @out;
ROW: for my $row (@{ $self->{rows} }) {
for my $col (keys %{$crit}) {
my $v = $crit->{$col};
id => $args{id} // 'entry',
_cols => $args{cols} // ['entry'],
_rows => $args{rows} // [],
_schema => $args{schema} // {},
_ts => $args{updated} // 1,
}, $class;
}
sub columns { return $_[0]->{_cols} }
sub schema { return $_[0]->{_schema} }
sub updated { return $_[0]->{_ts} }
sub set_logger { $_[0]->{_logger} = $_[1]; return $_[0] }
sub selectall_arrayref {
my ($self, $criteria) = @_;
my @rows = @{ $self->{_rows} };
for my $col (keys %{ $criteria // {} }) {
my $val = $criteria->{$col};
if (ref($val) eq 'HASH') {
for my $op (keys %{$val}) {
my $v = $val->{$op};
if ($op eq '>') { @rows = grep { defined $_->{$col} && $_->{$col} > $v } @rows }
( run in 0.558 second using v1.01-cache-2.11-cpan-7da0f45c409 )