Module: Neo4jBolt

Extended by:

Neo4jBolt

Included in:

Neo4jBolt

Defined in:

lib/neo4j_bolt.rb,
lib/neo4j_bolt/version.rb

Defined Under Namespace

Modules: DriverRegistry Classes: ConstraintValidationFailedError, Error, ExpectedOneResultError, IntegerOutOfRangeError, Node, Relationship, SyntaxError, TransactionContext

Constant Summary

CONSTRAINT_INDEX_PREFIX =

"neo4j_bolt_"

MIN_INTEGER =

-(2**63)

MAX_INTEGER =

(2**63) - 1

TRANSACTION_CONTEXT_KEY =

:neo4j_bolt_transaction_contexts

VERSION =

"0.4.0"

Class Attribute Summary

• .bolt_host ⇒ Object

  Returns the value of attribute bolt_host.

• .bolt_port ⇒ Object

  Returns the value of attribute bolt_port.

• .bolt_verbosity ⇒ Object

  Returns the value of attribute bolt_verbosity.

Instance Method Summary

• #cleanup_neo4j ⇒ Object
• #dump_database(io) ⇒ Object
• #load_database_dump(io, force_append: false) ⇒ Object
• #neo4j_query(query, data = {}, &block) ⇒ Object
• #neo4j_query_expect_one(query, data = {}) ⇒ Object
• #rollback ⇒ Object

  Explicit rollback is rollback-only: the surrounding block may finish, but its one real upstream transaction can no longer commit.

• #setup_constraints_and_indexes(constraints, indexes) ⇒ Object
• #transaction ⇒ Object
• #wait_for_neo4j ⇒ Object

Class Attribute Details

.bolt_host ⇒ Object

Returns the value of attribute bolt_host.

# File 'lib/neo4j_bolt.rb', line 140

def bolt_host
  @bolt_host
end

.bolt_port ⇒ Object

Returns the value of attribute bolt_port.

# File 'lib/neo4j_bolt.rb', line 140

def bolt_port
  @bolt_port
end

.bolt_verbosity ⇒ Object

Returns the value of attribute bolt_verbosity.

# File 'lib/neo4j_bolt.rb', line 141

def bolt_verbosity
  @bolt_verbosity
end

Instance Method Details

#cleanup_neo4j ⇒ Object

# File 'lib/neo4j_bolt.rb', line 254

def cleanup_neo4j
  DriverRegistry.cleanup
  nil
rescue Neo4j::Driver::Exceptions::Neo4jException => error
  raise_mapped_error(error)
end

#dump_database(io) ⇒ Object

# File 'lib/neo4j_bolt.rb', line 322

def dump_database(io)
  dump_ids = {}
  next_dump_id = 0
  identity_function = dump_identity_function

  neo4j_query("MATCH (n) RETURN n ORDER BY #{identity_function}(n)") do |row|
    node = row["n"]
    dump_ids[node.element_id || node.id.to_s] = next_dump_id
    io.puts "n #{JSON.generate(id: next_dump_id, labels: node.labels, properties: node)}"
    next_dump_id += 1
  end

  neo4j_query("MATCH ()-[r]->() RETURN r ORDER BY #{identity_function}(r)") do |row|
    relationship = row["r"]
    from_identity = relationship.start_node_element_id || relationship.start_node_id.to_s
    to_identity = relationship.end_node_element_id || relationship.end_node_id.to_s
    io.puts "r #{JSON.generate(
      from: dump_ids.fetch(from_identity),
      to: dump_ids.fetch(to_identity),
      type: relationship.type,
      properties: relationship
    )}"
  end
  nil
end

#load_database_dump(io, force_append: false) ⇒ Object

# File 'lib/neo4j_bolt.rb', line 348

def load_database_dump(io, force_append: false)
  unless force_append
    count = neo4j_query_expect_one("MATCH (n) RETURN count(n) AS count")["count"]
    raise Error, "There are nodes in this database, exiting now." unless count.zero?
  end

  node_batches = Hash.new { |hash, key| hash[key] = [] }
  relationship_batches = Hash.new { |hash, key| hash[key] = [] }
  parse_dump(io, node_batches, relationship_batches)

  temporary_id_property = "__neo4j_bolt_load_#{SecureRandom.hex(12)}"
  quoted_temporary_id = quote_identifier(temporary_id_property)
  loaded_nodes = 0
  loaded_relationships = 0

  begin
    node_batches.each_value do |nodes|
      nodes.each_slice(256) do |slice|
        labels = slice.first.fetch("labels")
        label_clause = labels.map { |label| ":#{quote_identifier(label)}" }.join
        count = neo4j_query_expect_one(<<~CYPHER, nodes: slice)["count_n"]
          UNWIND $nodes AS item
          CREATE (n#{label_clause})
          SET n = item.properties
          SET n.#{quoted_temporary_id} = item.id
          RETURN count(n) AS count_n
        CYPHER
        raise Error, "Expected #{slice.size} nodes, got #{count}." unless count == slice.size

        loaded_nodes += slice.size
        report_load_progress(loaded_nodes, loaded_relationships)
      end
    end

    relationship_batches.each do |type, relationships|
      relationships.each_slice(256) do |slice|
        count = neo4j_query_expect_one(<<~CYPHER, relationships: slice)["count_r"]
          UNWIND $relationships AS item
          MATCH (from) WHERE from.#{quoted_temporary_id} = item.from
          MATCH (to) WHERE to.#{quoted_temporary_id} = item.to
          CREATE (from)-[r:#{quote_identifier(type)}]->(to)
          SET r = item.properties
          RETURN count(r) AS count_r
        CYPHER
        raise Error, "Expected #{slice.size} relationships, got #{count}." unless count == slice.size

        loaded_relationships += slice.size
        report_load_progress(loaded_nodes, loaded_relationships)
      end
    end
  ensure
    original_error = $!
    begin
      neo4j_query("MATCH (n) WHERE n.#{quoted_temporary_id} IS NOT NULL REMOVE n.#{quoted_temporary_id}") if loaded_nodes.positive?
    rescue Error
      raise if original_error.nil?
    end
  end

  warn if loaded_nodes.positive? || loaded_relationships.positive?
  nil
end

#neo4j_query(query, data = {}, &block) ⇒ Object

# File 'lib/neo4j_bolt.rb', line 167

def neo4j_query(query, data = {}, &block)
  validate_parameters!(data)
  log_query(query, data)

  context = transaction_context
  if context
    run_and_convert(context.transaction, query, data, &block)
  else
    DriverRegistry.with_driver do |driver|
      driver.session do |session|
        run_and_convert(session, query, data, &block)
      end
    end
  end
rescue Neo4j::Driver::Exceptions::Neo4jException => error
  mark_transaction_failed!
  raise_mapped_error(error)
rescue IntegerOutOfRangeError
  mark_transaction_failed!
  raise
rescue Exception # rubocop:disable Lint/RescueException -- preserve rollback-only on callback/local errors
  mark_transaction_failed!
  raise
end

#neo4j_query_expect_one(query, data = {}) ⇒ Object

Raises:

• (ExpectedOneResultError)

# File 'lib/neo4j_bolt.rb', line 192

def neo4j_query_expect_one(query, data = {})
  rows = neo4j_query(query, data)
  return rows.first if rows.size == 1

  mark_transaction_failed!
  raise ExpectedOneResultError, "Expected one result, but got #{rows.size}."
end

#rollback ⇒ Object

Explicit rollback is rollback-only: the surrounding block may finish, but its one real upstream transaction can no longer commit.

Raises:

• (Error)

# File 'lib/neo4j_bolt.rb', line 246

def rollback
  context = transaction_context
  raise Error, "rollback called outside a transaction" unless context

  context.rollback_only = true
  nil
end

#setup_constraints_and_indexes(constraints, indexes) ⇒ Object

# File 'lib/neo4j_bolt.rb', line 280

def setup_constraints_and_indexes(constraints, indexes)
  wanted_constraints = Set.new
  wanted_indexes = Set.new

  constraints.each do |entry|
    label, property = schema_entry(entry, "constraint")
    name = "#{CONSTRAINT_INDEX_PREFIX}#{label}_#{property}"
    wanted_constraints << name
    neo4j_query(
      "CREATE CONSTRAINT #{quote_identifier(name)} IF NOT EXISTS " \
      "FOR (n:#{quote_identifier(label)}) REQUIRE n.#{quote_identifier(property)} IS UNIQUE"
    )
  end

  indexes.each do |entry|
    label, property = schema_entry(entry, "index")
    name = "#{CONSTRAINT_INDEX_PREFIX}#{label}_#{property}"
    wanted_indexes << name
    neo4j_query(
      "CREATE INDEX #{quote_identifier(name)} IF NOT EXISTS " \
      "FOR (n:#{quote_identifier(label)}) ON (n.#{quote_identifier(property)})"
    )
  end

  neo4j_query("SHOW CONSTRAINTS").each do |row|
    name = row["name"]
    next unless name&.start_with?(CONSTRAINT_INDEX_PREFIX)
    next if wanted_constraints.include?(name)

    neo4j_query("DROP CONSTRAINT #{quote_identifier(name)}")
  end

  neo4j_query("SHOW INDEXES").each do |row|
    name = row["name"]
    next unless name&.start_with?(CONSTRAINT_INDEX_PREFIX)
    next if wanted_indexes.include?(name) || wanted_constraints.include?(name)

    neo4j_query("DROP INDEX #{quote_identifier(name)}")
  end
  nil
end

#transaction ⇒ Object

# File 'lib/neo4j_bolt.rb', line 200

def transaction
  context = transaction_context
  if context
    context.depth += 1
    begin
      yield
    rescue Exception # rubocop:disable Lint/RescueException -- rollback-only must survive every unwind
      context.rollback_only = true
      raise
    ensure
      context.depth -= 1
    end
    return nil
  end

  DriverRegistry.with_driver do |driver|
    driver.session do |session|
      upstream_transaction = session.begin_transaction
      context = TransactionContext.new(
        session: session, transaction: upstream_transaction, depth: 1, rollback_only: false
      )
      transaction_contexts[self.object_id] = context

      begin
        yield
      rescue Exception # rubocop:disable Lint/RescueException -- ensure rollback for non-StandardError too
        context.rollback_only = true
        raise
      ensure
        begin
          context.rollback_only ? upstream_transaction.rollback : upstream_transaction.commit
        rescue Neo4j::Driver::Exceptions::Neo4jException => error
          raise_mapped_error(error)
        ensure
          transaction_contexts.delete(self.object_id)
        end
      end
    end
  end
  nil
rescue Neo4j::Driver::Exceptions::Neo4jException => error
  raise_mapped_error(error)
end

#wait_for_neo4j ⇒ Object

# File 'lib/neo4j_bolt.rb', line 261

def wait_for_neo4j
  attempts = Integer(ENV.fetch("NEO4J_BOLT_WAIT_ATTEMPTS", "30"))
  delay = Float(ENV.fetch("NEO4J_BOLT_WAIT_DELAY", "1"))
  last_error = nil

  attempts.times do |attempt|
    begin
      neo4j_query("RETURN 1 AS ready")
      return true
    rescue Error => error
      last_error = error
      warn "Waiting for Neo4j (attempt #{attempt + 1}/#{attempts})..." if Neo4jBolt.bolt_verbosity.to_i.positive?
      sleep delay if attempt + 1 < attempts
    end
  end

  raise(last_error || Error.new("Neo4j did not become ready"))
end