Class: ActiveRecordSpannerAdapter::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/activerecord_spanner_adapter/connection.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ Connection

Returns a new instance of Connection.



16
17
18
19
20
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 16

def initialize config
  @instance_id = config[:instance]
  @database_id = config[:database]
  @spanner = self.class.spanners config
end

Instance Attribute Details

#current_transaction ⇒ Object

Returns the value of attribute current_transaction.



14
15
16
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 14

def current_transaction
  @current_transaction
end

#database_id ⇒ Object (readonly)

Returns the value of attribute database_id.



13
14
15
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 13

def database_id
  @database_id
end

#instance_id ⇒ Object (readonly)

Returns the value of attribute instance_id.



13
14
15
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 13

def instance_id
  @instance_id
end

#spanner ⇒ Object (readonly)

Returns the value of attribute spanner.



13
14
15
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 13

def spanner
  @spanner
end

Class Method Details

.database_path(config) ⇒ Object 



293
294
295
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 293

def self.database_path config
  "#{config[:emulator_host]}/#{config[:project]}/#{config[:instance]}/#{config[:database]}"
end

.information_schema(config) ⇒ Object 



46
47
48
49
50
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 46

def self.information_schema config
  @information_schemas ||= {}
  @information_schemas[database_path(config)] ||= \
    ActiveRecordSpannerAdapter::InformationSchema.new new(config)
end

.reset_information_schemas! ⇒ Object

Clears the cached information about the underlying information schemas. Call this method if you drop and recreate a database with the same name to prevent the cached information to be used for the new database.



42
43
44
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 42

def self.reset_information_schemas!
  @information_schemas = {}
end

.spanners(config) ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 22

def self.spanners config
  config = config.symbolize_keys
  @spanners ||= {}
  @mutex ||= Mutex.new
  @mutex.synchronize do
    @spanners[database_path(config)] ||= Google::Cloud::Spanner.new(
      project_id: config[:project],
      credentials: config[:credentials],
      emulator_host: config[:emulator_host],
      scope: config[:scope],
      timeout: config[:timeout],
      lib_name: "spanner-activerecord-adapter",
      lib_version: ActiveRecordSpannerAdapter::VERSION
    )
  end
end

Instance Method Details

#abort_batch ⇒ Object

Aborts the current batch on this connection. This is a no-op if there is no batch on this connection.

See Also:



179
180
181
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 179

def abort_batch
  @ddl_batch = nil
end

#active? ⇒ Boolean

Returns:

  • (Boolean) 


58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 58

def active?
  # This method should not initialize a session.
  unless @session
    return false
  end
  # Assume that it is still active if it has been used in the past 50 minutes.
  if ((Time.current - @last_used) / 60).round < 50
    return true
  end
  session.execute_query "SELECT 1"
  true
rescue StandardError
  false
end

#begin_transaction(isolation = nil) ⇒ Object

Transactions



268
269
270
271
272
273
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 268

def begin_transaction isolation = nil
  raise "Nested transactions are not allowed" if current_transaction&.active?
  self.current_transaction = Transaction.new self, isolation
  current_transaction.begin
  current_transaction
end

#commit_transaction ⇒ Object 



275
276
277
278
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 275

def commit_transaction
  raise "This connection does not have a transaction" unless current_transaction
  current_transaction.commit
end

#create_database ⇒ Object

Database Operations



88
89
90
91
92
93
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 88

def create_database
  job = spanner.create_database instance_id, database_id
  job.wait_until_done!
  raise Google::Cloud::Error.from_error job.error if job.error?
  job.database
end

#create_transaction_after_failed_first_statement(original_error) ⇒ Object

Creates a transaction using a BeginTransaction RPC. This is used if the first statement of a transaction fails, as that also means that no transaction id was returned.



258
259
260
261
262
263
264
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 258

def create_transaction_after_failed_first_statement original_error
  transaction = current_transaction.force_begin_read_write
  Google::Cloud::Spanner::V1::TransactionSelector.new id: transaction.transaction_id
rescue Google::Cloud::Error
  # Raise the original error if the BeginTransaction RPC also fails.
  raise original_error
end

#database ⇒ Object 



95
96
97
98
99
100
101
102
103
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 95

def database
  @database ||= begin
    database = spanner.database instance_id, database_id
    unless database
      raise ActiveRecord::NoDatabaseError, "#{spanner.project}/#{instance_id}/#{database_id}"
    end
    database
  end
end

#ddl_batch ⇒ Object

Executes a set of DDL statements as one batch. This method raises an error if no block is given.

Examples:

connection.ddl_batch do
  connection.execute_ddl "CREATE TABLE `Users` (Id INT64, Name STRING(MAX)) PRIMARY KEY (Id)"
  connection.execute_ddl "CREATE INDEX Idx_Users_Name ON `Users` (Name)"
end

Raises:

  • (Google::Cloud::FailedPreconditionError) 


134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 134

def ddl_batch
  raise Google::Cloud::FailedPreconditionError, "No block given for the DDL batch" unless block_given?
  begin
    start_batch_ddl
    yield
    run_batch
  rescue StandardError
    abort_batch
    raise
  ensure
    @ddl_batch = nil
  end
end

#ddl_batch? ⇒ Boolean

Returns true if this connection is currently executing a DDL batch, and otherwise false.

Returns:

  • (Boolean) 


150
151
152
153
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 150

def ddl_batch?
  return true if @ddl_batch
  false
end

#disconnect! ⇒ Object 



73
74
75
76
77
78
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 73

def disconnect!
  session.release!
  true
ensure
  @session = nil
end

#execute_ddl(statements, operation_id: nil, wait_until_done: true) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 108

def execute_ddl statements, operation_id: nil, wait_until_done: true
  raise "DDL cannot be executed during a transaction" if current_transaction&.active?
  self.current_transaction = nil

  statements = Array statements
  return unless statements.any?

  # If a DDL batch is active we only buffer the statements on the connection until the batch is run.
  if @ddl_batch
    @ddl_batch.push(*statements)
    return true
  end

  execute_ddl_statements statements, operation_id, wait_until_done
end

#execute_query(sql, params: nil, types: nil, single_use_selector: nil, request_options: nil) ⇒ Object

DQL, DML Statements



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 203

def execute_query sql, params: nil, types: nil, single_use_selector: nil, request_options: nil
  if params
    converted_params, types = \
      Google::Cloud::Spanner::Convert.to_input_params_and_types(
        params, types
      )
  end

  # Clear the transaction from the previous statement.
  unless current_transaction&.active?
    self.current_transaction = nil
  end

  selector = transaction_selector || single_use_selector
  execute_sql_request sql, converted_params, types, selector, request_options
end

#execute_sql_request(sql, converted_params, types, selector, request_options = nil) ⇒ Object 



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 220

def execute_sql_request sql, converted_params, types, selector, request_options = nil
  res = session.execute_query \
    sql,
    params: converted_params,
    types: types,
    transaction: selector,
    request_options: request_options,
    seqno: (current_transaction&.next_sequence_number)
  current_transaction.grpc_transaction = res.metadata.transaction \
      if current_transaction && res&.metadata&.transaction
  res
rescue Google::Cloud::AbortedError
  # Mark the current transaction as aborted to prevent any unnecessary further requests on the transaction.
  current_transaction&.mark_aborted
  raise
rescue Google::Cloud::NotFoundError => e
  if session_not_found?(e) || transaction_not_found?(e)
    reset!
    # Force a retry of the entire transaction if this statement was executed as part of a transaction.
    # Otherwise, just retry the statement itself.
    raise_aborted_err if current_transaction&.active?
    retry
  end
  raise
rescue Google::Cloud::Error => e
  # Check if it was the first statement in a transaction that included a BeginTransaction
  # option in the request. If so, execute an explicit BeginTransaction and then retry the
  # request without the BeginTransaction option.
  if current_transaction && selector&.begin&.read_write
    selector = create_transaction_after_failed_first_statement e
    retry
  end
  # It was not the first statement, so propagate the error.
  raise
end

#raise_aborted_err ⇒ Object 



315
316
317
318
319
320
321
322
323
324
325
326
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 315

def raise_aborted_err
  retry_info = Google::Rpc::RetryInfo.new retry_delay: Google::Protobuf::Duration.new(seconds: 0, nanos: 1)
  begin
    raise GRPC::BadStatus.new(
      GRPC::Core::StatusCodes::ABORTED,
      "Transaction aborted",
      "google.rpc.retryinfo-bin": Google::Rpc::RetryInfo.encode(retry_info)
    )
  rescue GRPC::BadStatus
    raise Google::Cloud::AbortedError
  end
end

#reset! ⇒ Object 



80
81
82
83
84
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 80

def reset!
  disconnect!
  session
  true
end

#rollback_transaction ⇒ Object 



280
281
282
283
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 280

def rollback_transaction
  raise "This connection does not have a transaction" unless current_transaction
  current_transaction.rollback
end

#run_batch ⇒ Object

Runs the current batch on this connection. This will raise a FailedPreconditionError if there is no active batch on this connection.

See Also:



188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 188

def run_batch
  unless @ddl_batch
    raise Google::Cloud::FailedPreconditionError, "There is no batch active on this connection"
  end
  # Just return if the batch is empty.
  return true if @ddl_batch.empty?
  begin
    execute_ddl_statements @ddl_batch, nil, true
  ensure
    @ddl_batch = nil
  end
end

#session ⇒ Object Also known as: connect! 



52
53
54
55
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 52

def session
  @last_used = Time.current
  @session ||= spanner.create_session instance_id, database_id
end

#session_not_found?(err) ⇒ Boolean

Returns:

  • (Boolean) 


297
298
299
300
301
302
303
304
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 297

def session_not_found? err
  if err.respond_to?(:metadata) && err.metadata["google.rpc.resourceinfo-bin"]
    resource_info = Google::Rpc::ResourceInfo.decode err.metadata["google.rpc.resourceinfo-bin"]
    type = resource_info["resource_type"]
    return "type.googleapis.com/google.spanner.v1.Session".eql? type
  end
  false
end

#start_batch_ddl ⇒ Object

Starts a manual DDL batch. The batch must be ended by calling either run_batch or abort_batch.

Examples:

begin
  connection.start_batch_ddl
  connection.execute_ddl "CREATE TABLE `Users` (Id INT64, Name STRING(MAX)) PRIMARY KEY (Id)"
  connection.execute_ddl "CREATE INDEX Idx_Users_Name ON `Users` (Name)"
  connection.run_batch
rescue StandardError
  connection.abort_batch
  raise
end


168
169
170
171
172
173
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 168

def start_batch_ddl
  if @ddl_batch
    raise Google::Cloud::FailedPreconditionError, "A DDL batch is already active on this connection"
  end
  @ddl_batch = []
end

#transaction_not_found?(err) ⇒ Boolean

Returns:

  • (Boolean) 


306
307
308
309
310
311
312
313
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 306

def transaction_not_found? err
  if err.respond_to?(:metadata) && err.metadata["google.rpc.resourceinfo-bin"]
    resource_info = Google::Rpc::ResourceInfo.decode err.metadata["google.rpc.resourceinfo-bin"]
    type = resource_info["resource_type"]
    return "type.googleapis.com/google.spanner.v1.Transaction".eql? type
  end
  false
end

#transaction_selector ⇒ Object 



285
286
287
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 285

def transaction_selector
  return current_transaction&.transaction_selector if current_transaction&.active?
end

#truncate(table_name) ⇒ Object 



289
290
291
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 289

def truncate table_name
  session.delete table_name
end