sub update results from the CPAN

sub update

DBIO

view release on metacpan or search on metacpan

docs/adr/0036-atomic-upsert.md view on Meta::CPAN

## Context

### The over-promise this ticket closes

The driver READMEs (Pg, MySQL, SQLite) advertised "upsert via
`insert_or_update`". That was never true. `DBIO::Row::insert_or_update`
(`lib/DBIO/Row.pm`) is a bare alias:

```perl
sub insert_or_update { shift->update_or_insert(@_) }
sub update_or_insert { my $self = shift; return ($self->in_storage ? $self->update : $self->insert); }
```

It is a *client-side, two-step* helper: it looks at the in-memory
`in_storage` flag and issues either an `UPDATE` or an `INSERT`. It never
consults the server about whether a conflicting row exists, so it is subject
to the classic time-of-check/time-of-use race. The same is true of the
ResultSet-level composite writes that read-then-write:
`find_or_create`, `update_or_create` (`lib/DBIO/ResultSet.pm`), each of which
already carries a `B<Note>` in its POD warning that it "is subject to a race
condition ... use ... inside a transaction". These are honest, useful methods

lib/DBIO/Admin.pm view on Meta::CPAN

  $set ||= $self->set;
  Carp::croak('insert requires --resultset/--class and --set') unless $rs && $set;

  my $resultset = $self->schema->resultset($rs);
  my $obj = $resultset->new_result($set)->insert;
  print ''.ref($resultset).' ID: '.join(',', $obj->id)."\n" if !$self->quiet;
  return $obj;
}


sub update {
  my ($self, $rs, $set, $where) = @_;

  $rs ||= $self->resultset;
  $set ||= $self->set;
  $where ||= $self->where || {};
  Carp::croak('update requires --resultset/--class and --set') unless $rs && $set;

  my $resultset = $self->schema->resultset($rs)->search($where);
  if (!$self->quiet) {
    my $count = eval { $resultset->count };

lib/DBIO/ChangeLog.pm view on Meta::CPAN


  if ($self->_changelog_is_tracked && !$self->_changelog_is_disabled) {
    my %cols = $self->_changelog_filtered_columns($self->get_columns);
    $self->_changelog_record('insert', \%cols);
  }

  return $result;
}


sub update {
  my ($self, $upd) = @_;

  # Apply the update values so get_dirty_columns sees them
  $self->set_inflated_columns($upd) if $upd;

  my %dirty = $self->get_dirty_columns;

  # Filter out excluded columns
  my $exclude = $self->_changelog_excluded_columns;
  delete $dirty{$_} for keys %$exclude;

lib/DBIO/Ordered.pm view on Meta::CPAN

        ? $self->_next_position_value($lsib_posval)
        : $self->_initial_position_value
      )
    );
  }

  return $self->next::method(@_);
}


sub update {
  my $self = shift;

  return $self->next::method(@_) if $self->result_source->schema->{_ORDERED_INTERNAL_UPDATE};

  my $upd = shift;
  $self->set_inflated_columns($upd) if $upd;

  my $position_column = $self->position_column;
  my @group_columns = $self->_grouping_columns;

lib/DBIO/Relationship/Base.pm view on Meta::CPAN

sub find_or_create_related {
  my ($self, $rel) = (shift, shift);

  my $obj = $self->related_resultset($rel)->find_or_create(@_);
  delete $self->{related_resultsets}{$rel};

  return $obj;
}


sub update_or_create_related {
  #my ($self, $rel, @args) = @_;
  shift->related_resultset(shift)->update_or_create(@_);
}


sub set_from_related {
  my ($self, $rel, $f_obj) = @_;

  $self->set_columns($self->_inferred_from_related($rel, $f_obj));

lib/DBIO/Relationship/Base.pm view on Meta::CPAN

    $self->throw_exception(
      "Relationship '$rel' does not resolve to a concrete column condition"
    . " (column(s) '@{[ join q{, }, @unresolved ]}' would be set to undef)"
    ) if @unresolved;
  }

  return $inferred || {};
}


sub update_from_related {
  my $self = shift;
  $self->set_from_related(@_);
  $self->update;
}


sub delete_related {
  my $self = shift;
  my $obj = $self->search_related(@_)->delete;
  delete $self->{related_resultsets}->{$_[0]};

lib/DBIO/Relationship/CascadeActions.pm view on Meta::CPAN

    }

    $guard->commit;
    return $ret;
  }

  $self->next::method(@rest);
}


sub update {
  my ($self, @rest) = @_;
  return $self->next::method(@rest) unless ref $self;
    # Because update cascades on a class *really* don't make sense!

  my $source = $self->result_source;
  my %rels = map { $_ => $source->relationship_info($_) } $source->relationships;
  my @cascade = grep { $rels{$_}{attrs}{cascade_update} } keys %rels;

  if (@cascade) {
    my $guard = $source->schema->txn_scope_guard;

lib/DBIO/ResultSet.pm view on Meta::CPAN

    $op eq 'update' ? $values : (),
    $cond,
  ) : '0E0';

  $guard->commit if $guard;

  return $res;
}


sub update {
  my ($self, $values, $options) = @_;
  $self->throw_exception('Values for update must be a hash')
    unless ref $values eq 'HASH';
  $self->_assert_restricted_or_all('update', $options);

  # no row is involved: every value goes to the storage as it is
  assert_column_value($_, $values->{$_}) for sort keys %$values;

  return $self->_rs_update_delete ('update', $values);
}


sub update_all {
  my ($self, $values, $options) = @_;
  $self->throw_exception('Values for update_all must be a hash')
    unless ref $values eq 'HASH';
  $self->_assert_restricted_or_all('update_all', $options);

  my $guard = $self->result_source->schema->txn_scope_guard;
  $_->update({%$values}) for $self->all;  # shallow copy - update will mangle it
  $guard->commit;
  return 1;
}

lib/DBIO/ResultSet.pm view on Meta::CPAN

        my ($row) = @_;
        return $row if $row;
        return $self->create_async($hash);
      });
    },
    sub { $self->find_or_create($hash, $attrs) },
  );
}


sub update_or_create {
  my $self = shift;
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $cond = ref $_[0] eq 'HASH' ? shift : {@_};

  # find() only accepts a condition it can reduce to concrete columns on this
  # table, so the lookup drops any relation-valued key that does not resolve to
  # a concrete foreign-key condition. The FULL condition -- relation keys
  # included -- still reaches the row layer via update()/insert below, which
  # owns nested-relation semantics.
  my @lookup_cols;

lib/DBIO/ResultSet.pm view on Meta::CPAN

      = $self->_resolve_related_cond($rsrc, $key, $lookup{$key});

    delete $lookup{$key}
      unless $self->_related_cond_is_concrete($rel_cond, $crosstable);
  }

  return \%lookup;
}


sub update_or_create_async {
  my $self  = shift;
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};

  return $self->_rs_run_async('update_or_create',
    sub {
      my ($backend, $fc) = @_;

      my @lookup_cols;
      return $self->_find_for_write_async($self->_find_lookup_cond($cond), $attrs, 'update_or_create', \@lookup_cols)->then(sub {

lib/DBIO/ResultSet.pm view on Meta::CPAN

      my $crow = $child->{rs}->new_result($child->{cols});
      $crow->set_from_related($_, $row) for @{ $after->{revs} };
      $self->_upsert_run($child, $crow);
    }
  }

  return $row;
}


sub update_or_new {
    my $self  = shift;
    my $attrs = ( @_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {} );
    my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};

    my @lookup_cols;
    my $row = $self->_find_for_write( $cond, $attrs, 'update_or_new', \@lookup_cols );
    if ( defined $row ) {
        $row->update($self->_update_data_for_found_row($cond, \@lookup_cols));
        return $row;
    }

    return $self->new_result($cond);
}


sub update_or_new_async {
  my $self  = shift;
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};

  return $self->_rs_run_async('update_or_new',
    sub {
      my ($backend, $fc) = @_;

      my @lookup_cols;
      return $self->_find_for_write_async($cond, $attrs, 'update_or_new', \@lookup_cols)->then(sub {

lib/DBIO/Row.pm view on Meta::CPAN

    if ($accessor eq 'single') {
      next unless exists $self->{_relationship_data}{$rel}
        or ($upd and exists $upd->{$rel} and (!defined $upd->{$rel} or ref $upd->{$rel}));
    }
    return 1;
  }
  return 0;
}


sub update_async {
  my ($self, $upd) = @_;

  return $self->_row_run_async('update',
    sub {
      my ($backend, $fc) = @_;

      # ADR 0071: the hash is an assignment of the caller's
      $self->__assert_update_assignable($upd) if $upd;

      if (my $pkg = $self->_async_override_gap('update')) {

lib/DBIO/Row.pm view on Meta::CPAN

        delete $self->{_column_data_in_storage};
        $self->store_storage_values if @{$self->storage_value_columns};
        return $self;
      });
    },
    sub { $self->update($upd) },
  );
}


sub update {
  my ($self, $upd) = @_;

  # ADR 0071: the hash is an assignment of the caller's
  $self->__assert_update_assignable($upd) if $upd;

  # whatever a component has to do before the columns are set and written;
  # it may take keys out of $upd
  $self->prepare_update($upd);

  # ProxyResultSetMethod (integrated helper): strip proxy slots from the

lib/DBIO/Row.pm view on Meta::CPAN


  # StorageValues (integrated helper): snapshot freshly-inflated values.
  $new->store_storage_values if @{$new->storage_value_columns};

  return $new;
}


sub insert_or_update { shift->update_or_insert(@_) }

sub update_or_insert {
  my $self = shift;
  return ($self->in_storage ? $self->update : $self->insert);
}


sub insert_or_update_async { shift->update_or_insert_async(@_) }


sub update_or_insert_async {
  my $self = shift;
  return ($self->in_storage ? $self->update_async : $self->insert_async);
}


sub is_changed {
  return keys %{shift->{_dirty_columns} || {}};
}

lib/DBIO/Storage.pm view on Meta::CPAN

  qw/ quote_char name_sep quote_names /;
}


sub select { die "Virtual method!" }


sub insert { die "Virtual method!" }


sub update { die "Virtual method!" }


sub delete { die "Virtual method!" }


sub select_single { die "Virtual method!" }

# The force_pool value the read of a read-then-write call (find_or_create,
# find_or_new, update_or_create, update_or_new, the upsert fallback) carries in
# its attributes. Nothing by default; DBIO::Replicated::Storage answers 'master'.

lib/DBIO/Storage.pm view on Meta::CPAN


sub select_async { shift->_run_async('select', @_) }


sub select_single_async { shift->_run_async('select_single', @_) }


sub insert_async { shift->_run_async('insert', @_) }


sub update_async { shift->_run_async('update', @_) }


sub delete_async { shift->_run_async('delete', @_) }


sub txn_do_async { shift->_run_async('txn_do', @_) }


sub future_class {
  my $self = shift;

lib/DBIO/Storage/Async.pm view on Meta::CPAN

  return $self->_run_crud_pooled('select_single', @_);
}


sub insert_async {
  my $self = shift;
  return $self->_run_crud_pooled('insert', @_);
}


sub update_async {
  my $self = shift;
  return $self->_run_crud_pooled('update', @_);
}


sub delete_async {
  my $self = shift;
  return $self->_run_crud_pooled('delete', @_);
}

lib/DBIO/Storage/Async.pm view on Meta::CPAN

}

# --- Sync fallbacks ---
# Let sync methods work by blocking on the async result via ->get.
# Useful for scripts/migrations.


sub select        { my $self = shift; return $self->select_async(@_)->get        }
sub select_single { my $self = shift; return $self->select_single_async(@_)->get }
sub insert        { my $self = shift; return $self->insert_async(@_)->get        }
sub update        { my $self = shift; return $self->update_async(@_)->get        }
sub delete        { my $self = shift; return $self->delete_async(@_)->get        }
sub txn_do        { my $self = shift; return $self->txn_do_async(@_)->get        }
sub deploy        { my $self = shift; return $self->deploy_async(@_)->get        }

# --- Schema integration ---

sub schema { $_[0]->{schema} }


sub debug  { $_[0]->{debug} }

lib/DBIO/Storage/Async/PipelineContext.pm view on Meta::CPAN

}

sub insert_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('insert', $self->{pipeline_conn}, @_)
  );
}

sub update_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('update', $self->{pipeline_conn}, @_)
  );
}

sub delete_async {
  my $self = shift;
  $self->_assert_context_active;

lib/DBIO/Storage/Async/PipelineContext.pm view on Meta::CPAN

# statement's result arrives at _pipeline_sync, and the sync only runs once the
# batch coderef has returned. ->get inside the batch would therefore wait for a
# result whose delivery is waiting on the caller it is blocking. Refusing with
# the reason beats hanging the process (and beats "Can't locate object method",
# which reads like the context is simply incomplete).


sub select        { $_[0]->_no_blocking('select')        }
sub select_single { $_[0]->_no_blocking('select_single') }
sub insert        { $_[0]->_no_blocking('insert')        }
sub update        { $_[0]->_no_blocking('update')        }
sub delete        { $_[0]->_no_blocking('delete')        }

sub _no_blocking {
  my ($self, $op) = @_;
  croak "Blocking ->$op is not available inside a pipeline batch: a batched "
    . "statement resolves at the pipeline sync, which runs after the batch "
    . "coderef returns -- blocking on it here would deadlock. Call "
    . "->${op}_async and return an aggregate Future from the batch.";
}

lib/DBIO/Storage/Async/TransactionContext.pm view on Meta::CPAN

}

sub insert_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('insert', $self->{txn_conn}, @_)
  );
}

sub update_async {
  my $self = shift;
  $self->_assert_context_active;
  return $self->_track_statement(
    $self->{storage}->_run_crud_pinned('update', $self->{txn_conn}, @_)
  );
}

sub delete_async {
  my $self = shift;
  $self->_assert_context_active;

lib/DBIO/Storage/Async/TransactionContext.pm view on Meta::CPAN

    $self->{storage}->_run_crud_pinned('delete', $self->{txn_conn}, @_)
  );
}

# Sync fallbacks -- block the event loop via ->get


sub select             { my $self = shift; $self->select_async(@_)->get        }
sub select_single      { my $self = shift; $self->select_single_async(@_)->get }
sub insert             { my $self = shift; $self->insert_async(@_)->get        }
sub update             { my $self = shift; $self->update_async(@_)->get        }
sub delete             { my $self = shift; $self->delete_async(@_)->get        }

# Delegations


sub sql_maker    { my $self = shift; $self->{storage}->sql_maker(@_)    }
sub debug        { my $self = shift; $self->{storage}->debug(@_)        }

sub pipeline {
  my ($self, $coderef) = @_;

lib/DBIO/Storage/DBI.pm view on Meta::CPAN


  # the statements carry no bound value; the hook is asked all the same, a
  # driver's own rules apply to the text of the engine
  $self->throw_exception(
    ref $err ? $err : $self->_mask_error_values($err, [], undef)
  ) if defined $err;

  return $count;
}

sub update {
  #my ($self, $source, @args) = @_;
  shift->_execute('update', @_);
}


sub delete {
  #my ($self, $source, @args) = @_;
  shift->_execute('delete', @_);
}

t/composed/async_layer_core_read_capability.t view on Meta::CPAN

# ---------------------------------------------------------------------------
{
  package T::K268::ModelA;
  use base 'DBIO::Storage::Async';
  use mro 'c3';
  our @SELECTED;
  sub future_class        { 'DBIO::Future::Immediate' }
  sub select_async        { push @SELECTED, $_[1]; DBIO::Future::Immediate->done([ 1, 'row' ]) }
  sub select_single_async { DBIO::Future::Immediate->done([]) }
  sub insert_async        { DBIO::Future::Immediate->done([]) }
  sub update_async        { DBIO::Future::Immediate->done(0)  }
  sub delete_async        { DBIO::Future::Immediate->done(0)  }
  sub txn_do_async        { DBIO::Future::Immediate->done(1)  }
}

# Model A, taking connect options beside a broker: DBIO::Forked since k12.
{
  package T::K268::ModelA::Declaring;
  use base 'T::K268::ModelA';
  use mro 'c3';
  sub transport_capabilities { ('broker_connect_options') }

t/composed/async_layer_default_seam.t view on Meta::CPAN

# directly and runs the driver's ordinary sync CRUD elsewhere, so it inherits the
# abstract _query_async stub and there is NO seam for a layer to wrap.
{
  package T::LDS::ModelATransport;
  use base 'DBIO::Storage::Async';
  use mro 'c3';
  sub future_class        { 'DBIO::Future::Immediate' }
  sub select_async        { DBIO::Future::Immediate->done([]) }
  sub select_single_async { DBIO::Future::Immediate->done([]) }
  sub insert_async        { DBIO::Future::Immediate->done([]) }
  sub update_async        { DBIO::Future::Immediate->done(0)  }
  sub delete_async        { DBIO::Future::Immediate->done(0)  }
  sub txn_do_async        { DBIO::Future::Immediate->done(1)  }
}

# Sanity: the two transports really do differ in the one property under test.
my $abstract_seam = DBIO::Storage::Async->can('_query_async');
isnt( T::LDS::SeamTransport->can('_query_async'), $abstract_seam,
  'sanity: the Model B transport overrides the _query_async seam' );
is( T::LDS::ModelATransport->can('_query_async'), $abstract_seam,
  'sanity: the Model A transport inherits the abstract stub -- no seam' );

t/debug/async_upsert_trace.t view on Meta::CPAN

    if ($on_conflict) {
      $sql .= ' ON CONFLICT (id) DO UPDATE SET name = ?';
      push @bind, $self->{bindtype} eq 'columns'
        ? [ { dbic_colname => 'name' } => 'tail-value' ]
        : 'tail-value';
    }
    return ($sql, @bind);
  }

  # update and delete: a key of the options is consumed
  sub update {
    my ($self, $table, $set, $where, $options) = @_;
    delete $options->{marker} if ref $options eq 'HASH';
    return $self->next::method($table, $set, $where);
  }

  sub delete {
    my ($self, $table, $where, $options) = @_;
    delete $options->{marker} if ref $options eq 'HASH';
    return $self->next::method($table, $where);
  }

t/resultset/async_parity_helpers.t view on Meta::CPAN

    return DBIO::Future::Immediate->done( @row ? [ @row ] : undef );
  }

  sub insert_async {
    my ($self, @args) = @_;
    push @{$self->{calls}}, 'insert';
    my $cols = $self->_sync->insert(@args);
    return DBIO::Future::Immediate->done($cols);
  }

  sub update_async {
    my ($self, @args) = @_;
    push @{$self->{calls}}, 'update';
    my $rows = $self->_sync->update(@args);
    return DBIO::Future::Immediate->done($rows);
  }

  sub delete_async {
    my ($self, @args) = @_;
    push @{$self->{calls}}, 'delete';
    my $rows = $self->_sync->delete(@args);

t/resultset/update_or_create_related.t view on Meta::CPAN


  sub find       { my $self = shift; push @find_calls, [@_];       return $find_returns }
  sub new_result { my $self = shift; push @new_result_calls, [@_]; return TestRecordingResultSet::FakeRow->new }

  package TestRecordingResultSet::FakeRow;

  our @update_calls;

  sub new    { bless {}, shift }
  sub insert { $_[0] }
  sub update { my $self = shift; push @update_calls, [@_]; $self }
}

my $cond = {
  artistid => 1,
  name     => 'Updated Name',
  cds      => { title => 'New Title', year => 2020 },
};
my $lookup = { artistid => 1 }; # the constraint's columns only (ADR 0070); name is data, cds non-concrete

# (b1) row found -> $row->update() gets the full condition unsplit

t/row/async_overrides.t view on Meta::CPAN


  __PACKAGE__->table('own');
  __PACKAGE__->add_columns(
    id   => { data_type => 'integer', is_auto_increment => 1 },
    name => { data_type => 'varchar', size => 50 },
  );
  __PACKAGE__->set_primary_key('id');

  our @CALLS;
  sub insert { my $self = shift; push @CALLS, 'insert'; $self->name(uc $self->name); $self->next::method(@_) }
  sub update { my $self = shift; push @CALLS, 'update'; $self->next::method(@_) }
  sub delete { my $self = shift; push @CALLS, 'delete'; $self->next::method(@_) }
}

# ---- a user class that carries its own *_async sibling --------------------
{
  package Test::Ov::Schema::Result::Sibling;
  use base 'DBIO::Core';

  __PACKAGE__->table('sibling');
  __PACKAGE__->add_columns(

t/test/14_async_backend.t view on Meta::CPAN

  sub disconnect {
    my ($self) = @_;
    $self->{disconnected}++;
    $self->{pool}->shutdown if $self->{pool};
    $self->{pool} = undef;
    return 1;
  }
  sub select_async        { push @{$_[0]{calls}}, 'select';        DBIO::Future::Immediate->done('A:select') }
  sub select_single_async { push @{$_[0]{calls}}, 'select_single'; DBIO::Future::Immediate->done('A:select_single') }
  sub insert_async        { push @{$_[0]{calls}}, 'insert';        DBIO::Future::Immediate->done('A:insert') }
  sub update_async        { push @{$_[0]{calls}}, 'update';        DBIO::Future::Immediate->done('A:update') }
  sub delete_async        { push @{$_[0]{calls}}, 'delete';        DBIO::Future::Immediate->done('A:delete') }
  sub txn_do_async        { push @{$_[0]{calls}}, 'txn_do';        DBIO::Future::Immediate->done('A:txn_do') }

  package My::Mock::AsyncB;
  use base 'DBIO::Storage::Async';
  sub new { my ($c, $s) = @_; bless { schema => $s, calls => [] }, $c }
  sub future_class { 'DBIO::Future::Immediate' }
  sub connect_info { my $s = shift; $s->{connect_info} = shift if @_; $s->{connect_info} }
  sub disconnect { 1 }
  sub select_async { push @{$_[0]{calls}}, 'select'; DBIO::Future::Immediate->done('B:select') }

( run in 5.337 seconds using v1.01-cache-2.11-cpan-145b269f2f3 )