def query(statement, options = nil)
      policy = create_policy(options, QueryPolicy, default_query_policy)
      new_policy = policy.clone

      nodes = @cluster.nodes
      if nodes.empty?
        raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Scan failed because cluster is empty.")
      end

      recordset = Recordset.new(policy.record_queue_size, nodes.length, :query)

      # Use a thread per node
      nodes.each do |node|
        Thread.new do
          Thread.current.abort_on_exception = true
          command = QueryCommand.new(node, new_policy, statement, recordset)
          begin
            execute_command(command)
          rescue => e
            Aerospike.logger.error(e.backtrace.join("\n")) unless e == QUERY_TERMINATED_EXCEPTION
            recordset.cancel(e)
          ensure
            recordset.thread_finished
          end
        end
      end

      recordset
    end
def query(statement, options = nil)
      policy = create_policy(options, QueryPolicy, default_query_policy)
      new_policy = policy.clone

      nodes = @cluster.nodes
      if nodes.empty?
        raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::SERVER_NOT_AVAILABLE, "Scan failed because cluster is empty.")
      end

      recordset = Recordset.new(policy.record_queue_size, nodes.length, :query)

      # Use a thread per node
      nodes.each do |node|
        Thread.new do
          Thread.current.abort_on_exception = true
          command = QueryCommand.new(node, new_policy, statement, recordset)
          begin
            execute_command(command)
          rescue => e
            Aerospike.logger.error(e.backtrace.join("\n")) unless e == QUERY_TERMINATED_EXCEPTION
            recordset.cancel(e)
          ensure
            recordset.thread_finished
          end
        end
      end

      recordset
    end
--------------------------------------------------------
 Query functions (Supported by Aerospike 3 servers only)
--------------------------------------------------------
 Query executes a query and returns a recordset.
 The query executor puts records on a channel from separate goroutines.
 The caller can concurrently pops records off the channel through the
 record channel.

 This method is only supported by Aerospike 3 servers.
 If the policy is nil, a default policy will be generated.