Merge pull request #126 from aerospike/CLIENT-2575

CLIENT-2575 Supports new queries and record size filter
aerospike · Dec 15, 2023 · f6d8e31 · f6d8e31
2 parents b3f25d9 + 7928e24
commit f6d8e31
Show file tree

Hide file tree

Showing 17 changed files with 259 additions and 546 deletions.
diff --git a/lib/aerospike/cdt/map_operation.rb b/lib/aerospike/cdt/map_operation.rb
@@ -116,7 +116,7 @@ def initialize(op_type, map_op, bin_name, *arguments, ctx: nil, return_type: nil
       def self.create(bin_name, order, ctx: nil)
         if !ctx || ctx.length == 0
           # If context not defined, the set order for top-level bin map.
-          self.set_policy(MapPolicy.new(order: order, flag: 0), bin_name)
+          self.set_policy(MapPolicy.new(order: order, flags: 0), bin_name)
         else
           MapOperation.new(Operation::CDT_MODIFY, SET_TYPE, bin_name, order[:attr], ctx: ctx, flag: order[:flag])
         end

diff --git a/lib/aerospike/client.rb b/lib/aerospike/client.rb
@@ -529,7 +529,7 @@ def execute_udf(key, package_name, function_name, args = [], options = nil)
     # This method is only supported by Aerospike 3 servers.
     # If the policy is nil, the default relevant policy will be used.
     def execute_udf_on_query(statement, package_name, function_name, function_args = [], options = nil)
-      policy = create_policy(options, QueryPolicy, default_query_policy)
+      policy = create_policy(options, WritePolicy, default_write_policy)
 
       nodes = @cluster.nodes
       if nodes.empty?
@@ -538,14 +538,12 @@ def execute_udf_on_query(statement, package_name, function_name, function_args =
 
       statement = statement.clone
       statement.set_aggregate_function(package_name, function_name, function_args, false)
-
       # Use a thread per node
       nodes.each do |node|
-        partitions = node.cluster.node_partitions(node, statement.namespace)
         Thread.new do
           Thread.current.abort_on_exception = true
           begin
-            command = QueryCommand.new(node, policy, statement, nil, partitions)
+            command = ServerCommand.new(@cluster, node, policy, statement, true, statement.task_id)
             execute_command(command)
           rescue => e
             Aerospike.logger.error(e)
@@ -701,7 +699,6 @@ def scan_node(node, namespace, set_name, bin_names = nil, options = nil)
     # If the policy is nil, the default relevant policy will be used.
     def query_partitions(partition_filter, statement, options = nil)
       policy = create_policy(options, QueryPolicy, default_query_policy)
-      new_policy = policy.clone
 
       nodes = @cluster.nodes
       if nodes.empty?

diff --git a/lib/aerospike/command/batch_direct_command.rb b/lib/aerospike/command/batch_direct_command.rb
@@ -60,7 +60,7 @@ def write_buffer
         operation_count = bin_names.length
       end
 
-      write_header(policy, read_attr, 0, 2, operation_count)
+      write_header_read(policy, read_attr, 0, 2, operation_count)
       write_field_string(batch.namespace, Aerospike::FieldType::NAMESPACE)
       write_field_header(byte_size, Aerospike::FieldType::DIGEST_RIPE_ARRAY)
 

diff --git a/lib/aerospike/command/batch_index_command.rb b/lib/aerospike/command/batch_index_command.rb
@@ -66,7 +66,7 @@ def write_buffer
         end
       end
       size_buffer
-      write_header(policy,read_attr | INFO1_BATCH, 0, field_count, 0)
+      write_header_read(policy, read_attr | INFO1_BATCH, 0, field_count, 0)
 
       write_predexp(@policy.predexp, predexp_size)