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.



23
24
25
26
27
28
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 23

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

Instance Attribute Details

#current_transaction ⇒ Object

Returns the value of attribute current_transaction.



20
21
22
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 20

def current_transaction
  @current_transaction
end

#database_id ⇒ Object (readonly)

Returns the value of attribute database_id.



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

def database_id
  @database_id
end

#instance_id ⇒ Object (readonly)

Returns the value of attribute instance_id.



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

def instance_id
  @instance_id
end

#isolation_level ⇒ Object

Returns the value of attribute isolation_level.



21
22
23
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 21

def isolation_level
  @isolation_level
end

#spanner ⇒ Object (readonly)

Returns the value of attribute spanner.



19
20
21
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 19

def spanner
  @spanner
end

Class Method Details

.database_path(config) ⇒ Object 



384
385
386
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 384

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

.information_schema(config) ⇒ Object 



63
64
65
66
67
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 63

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.



54
55
56
57
58
59
60
61
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 54

def self.reset_information_schemas!
  return unless @database

  @information_schemas.each_value do |info_schema|
    info_schema.connection.disconnect!
  end
  @information_schemas = {}
end

.spanners(config) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 30

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:



222
223
224
225
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 222

def abort_batch
  @ddl_batch = nil
  @dml_batch = nil
end

#active? ⇒ Boolean

Returns:

  • (Boolean) 


75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 75

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, **options) ⇒ Object

Transactions



357
358
359
360
361
362
363
364
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 357

def begin_transaction isolation = nil, **options
  raise "Nested transactions are not allowed" if current_transaction&.active?
  exclude_from_streams = options.fetch :exclude_txn_from_change_streams, false
  self.current_transaction = Transaction.new self, isolation || @isolation_level,
                                             exclude_txn_from_change_streams: exclude_from_streams
  current_transaction.begin
  current_transaction
end

#commit_transaction ⇒ Object 



366
367
368
369
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 366

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

#create_database ⇒ Object

Database Operations



105
106
107
108
109
110
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 105

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.



347
348
349
350
351
352
353
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 347

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 



112
113
114
115
116
117
118
119
120
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 112

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) 


151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 151

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) 


167
168
169
170
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 167

def ddl_batch?
  return true if @ddl_batch
  false
end

#disconnect! ⇒ Object 



90
91
92
93
94
95
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 90

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

#dml_batch ⇒ Object

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

Raises:

  • (Google::Cloud::FailedPreconditionError) 


268
269
270
271
272
273
274
275
276
277
278
279
280
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 268

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

#dml_batch? ⇒ Boolean

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

Returns:

  • (Boolean) 


173
174
175
176
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 173

def dml_batch?
  return true if @dml_batch
  false
end

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



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 125

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, statement_type: nil) ⇒ Object

DQL, DML Statements



284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 284

def execute_query sql, params: nil, types: nil, single_use_selector: nil, request_options: nil, statement_type: nil
  # Clear the transaction from the previous statement.
  unless current_transaction&.active?
    self.current_transaction = nil
  end
  if statement_type == :dml && dml_batch?
    @dml_batch.push({ sql: sql, params: params, types: types })
    return
  end
  if params
    converted_params, types =
      Google::Cloud::Spanner::Convert.to_input_params_and_types(
        params, types
      )
  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

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 305

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
  if TransactionMutationLimitExceededError.is_mutation_limit_error? e
    raise
  end
  # 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

#native_database_types ⇒ Object 



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

def native_database_types
  NATIVE_DATABASE_TYPES
end

#raise_aborted_err ⇒ Object 



406
407
408
409
410
411
412
413
414
415
416
417
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 406

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 



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

def reset!
  disconnect!
  session
  true
end

#rollback_transaction ⇒ Object 



371
372
373
374
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 371

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:



232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 232

def run_batch
  unless @ddl_batch || @dml_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? || @dml_batch&.empty?
  begin
    if @ddl_batch
      run_ddl_batch
    else
      run_dml_batch
    end
  end
end

#run_ddl_batch ⇒ Object 



247
248
249
250
251
252
253
254
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 247

def run_ddl_batch
  return true if @ddl_batch.empty?
  begin
    execute_ddl_statements @ddl_batch, nil, true
  ensure
    @ddl_batch = nil
  end
end

#run_dml_batch ⇒ Object 



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

def run_dml_batch
  return true if @dml_batch.empty?
  begin
    # Execute the DML statements in the batch.
    execute_dml_statements_in_batch @dml_batch
  ensure
    @dml_batch = nil
  end
end

#session ⇒ Object Also known as: connect! 



69
70
71
72
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 69

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

#session_not_found?(err) ⇒ Boolean

Returns:

  • (Boolean) 


388
389
390
391
392
393
394
395
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 388

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


191
192
193
194
195
196
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 191

def start_batch_ddl
  if @ddl_batch && @dml_batch
    raise Google::Cloud::FailedPreconditionError, "Batch is already active on this connection"
  end
  @ddl_batch = []
end

#start_batch_dml ⇒ Object

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

Examples:

begin
  connection.start_batch_dml
  connection.execute_query "insert into `Users` (Id, Name) VALUES (1, 'Test 1')"
  connection.execute_query "insert into `Users` (Id, Name) VALUES (2, 'Test 2')"
  connection.run_batch
rescue StandardError
  connection.abort_batch
  raise
end


211
212
213
214
215
216
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 211

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

#transaction_not_found?(err) ⇒ Boolean

Returns:

  • (Boolean) 


397
398
399
400
401
402
403
404
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 397

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 



376
377
378
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 376

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

#truncate(table_name) ⇒ Object 



380
381
382
 
# File 'lib/activerecord_spanner_adapter/connection.rb', line 380

def truncate table_name
  session.delete table_name
end