Client 2621 Support persistant maps

lib/aerospike/cdt/map_operation.rb

@@ -110,14 +110,23 @@ def initialize(op_type, map_op, bin_name, *arguments, ctx: nil, return_type: nil
         self
       end
 
-      ##
-      # Creates a map create operation.
-      # Server creates map at given context level.
-      def self.create(bin_name, order, ctx: nil)
-        if !ctx || ctx.length == 0
+      # Create map create operation.
+      # Server creates a map at the given context level.
+      #
+      # @param [String] bin_name The bin name.
+      # @param [Integer] order The map order.
+      # @param [Boolean] persist_index If true, persist map index. A map index improves lookup performance,
+      #   but requires more storage. A map index can be created for a top-level
+      #   ordered map only. Nested and unordered map indexes are not supported.
+      # @param [String] ctx Optional path to a nested map. If not defined, the top-level map is used.
+      def self.create(bin_name, order, persistent_index, ctx: nil)
+        if !ctx || ctx.empty?
           # If context not defined, the set order for top-level bin map.
-          self.set_policy(MapPolicy.new(order: order, flag: 0), bin_name)
+          attr = order[:attr]
+          attr += 0x10 if persistent_index
+          MapOperation.new(Operation::CDT_MODIFY, SET_TYPE, bin_name, attr, ctx: ctx, flag: order[:flag])
         else
+          # Create nested map. persistIndex does not apply here, so ignore it
           MapOperation.new(Operation::CDT_MODIFY, SET_TYPE, bin_name, order[:attr], ctx: ctx, flag: order[:flag])
         end
       end
@@ -128,7 +137,12 @@ def self.create(bin_name, order, ctx: nil)
       #
       # The required map policy attributes can be changed after the map is created.
       def self.set_policy(bin_name, policy, ctx: nil)
-        MapOperation.new(Operation::CDT_MODIFY, SET_TYPE, bin_name, policy.order[:attr], ctx: ctx)
+        attr = policy.attributes
+        # Remove persistIndex flag for nested maps.
+        if !ctx.nil? && !ctx.empty? && (attr & 0x10) != 0
+          attr &= ~0x10
+        end
+        MapOperation.new(Operation::CDT_MODIFY, SET_TYPE, bin_name, attr, ctx: ctx)
       end
 
       ##
@@ -635,7 +649,7 @@ def pack_bin_value
         args.unshift(return_type) if return_type
 
         Packer.use do |packer|
-          if @ctx != nil && @ctx.length > 0
+          if @ctx != nil && !@ctx.empty?
             packer.write_array_header(3)
             Value.of(0xff).pack(packer)
 
@@ -645,12 +659,12 @@ def pack_bin_value
             Value.of(@map_op).pack(packer)
           else
             packer.write_raw_short(@map_op)
-            if args.length > 0
+            if !args.empty?
               packer.write_array_header(args.length)
             end
           end
 
-          if args.length > 0
+          if !args.empty?
             args.each do |value|
               Value.of(value).pack(packer)
             end

lib/aerospike/cdt/map_policy.rb

@@ -17,10 +17,9 @@
 module Aerospike
   module CDT
     class MapPolicy
-      attr_accessor :order, :write_mode, :flags
-      attr_accessor :item_command, :items_command, :attributes
+      attr_accessor :order, :write_mode, :flags, :item_command, :items_command, :attributes, :persist_index
 
-      def initialize(order: nil, write_mode: nil, flags: nil)
+      def initialize(order: nil, write_mode: nil, persist_index: false, flags: nil)
         if write_mode && flags
           raise ArgumentError, "Use write mode for server versions < 4.3; use write flags for server versions >= 4.3."
         end
@@ -30,6 +29,10 @@ def initialize(order: nil, write_mode: nil, flags: nil)
         @flags = flags || MapWriteFlags::DEFAULT
         @attributes = order ? order[:attr] : 0
 
+        if @persist_index
+          @attributes |= 0x10
+        end
+
         case @write_mode
         when CDT::MapWriteMode::DEFAULT
           @item_command = CDT::MapOperation::PUT

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?

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)
 

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)