Class: Dataflow::Nodes::SqlQueryNode

Inherits:
ComputeNode show all
Defined in:
lib/dataflow/nodes/sql_query_node.rb

Overview

Transforms the dependency’s dataset to a SQL-compatible one.

Constant Summary

Constants included from SchemaMixin

SchemaMixin::SAMPLE_DATA_OUTPUT, SchemaMixin::SEPARATOR

Instance Method Summary collapse

Methods inherited from ComputeNode

#all_dependencies, #compute, #data_node, data_node_opts, #dependencies, dependency_opts, ensure_data_node_exists, ensure_dependencies, #execute_local_batch_computation, #execute_local_computation, #execution_valid?, #explain_update, #force_computing_lock_release!, #locked_for_computing?, #make_batch_params, #needs_automatic_recomputing?, #recompute, #schema, #set_defaults, #updated?, #updated_at, #updated_at=

Methods included from SchemaMixin

#infer_partial_schema, #infer_schema, #sample_data, #schema_inferrer

Methods included from Dataflow::Node

#all_dependencies, find, #metadata, #recompute, #required_by, #validate!

Instance Method Details

#computed_query ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/dataflow/nodes/sql_query_node.rb', line 24

def computed_query
  # 1. replace the current write dataset's name
  q = query.gsub('<node>', write_dataset_name)

  # 2. replace the dependencies' (read) dataset names
  q.gsub(/<[0-9]+>/) do |match|
    # [1..-2] will remove the 'less than' < and 'greater than' >
    dep_index = match[1..-2].to_i
    raise "Specified depependency #{match} does not exist. There are only #{dependencies.count} dependencies." if dep_index >= dependencies.count
    dependencies[dep_index].read_dataset_name
  end
end

#execute_query ⇒ Object 



37
38
39
40
41
 
# File 'lib/dataflow/nodes/sql_query_node.rb', line 37

def execute_query
  query = computed_query
  logger.log(query)
  data_node.send(:db_adapter).client[query].to_a
end

#valid_for_computation? ⇒ Boolean

Returns:

  • (Boolean) 


10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/dataflow/nodes/sql_query_node.rb', line 10

def valid_for_computation?
  unless (data_node&.db_backend.to_s =~ /sql/).present?
    errors.add(:db_backend, 'Must have a SQL based backend.')
  end

  begin
    computed_query
  rescue StandardError => e
    errors.add(:query, "Specified query has errors: #{e.message}")
  end

  super
end