Skip to content

Commit

Permalink
Merge pull request #129 from aerospike/2.30.0
Browse files Browse the repository at this point in the history
3.0.0
  • Loading branch information
vmsachin committed Dec 18, 2023
2 parents b3f25d9 + ad754fc commit b9c8544
Show file tree
Hide file tree
Showing 21 changed files with 374 additions and 559 deletions.
9 changes: 9 additions & 0 deletions CHANGELOG.md
Expand Up @@ -2,6 +2,15 @@

All notable changes to this project will be documented in this file.

## [3.0.0] 2023-12-15
Notice: This version of the client only supports Aerospike Server v6.0 and later. Some features will work for the older server versions.
- **new_features**
- [CLIENT-2575] - Support Exp.recordSize().
- [CLIENT-2621] - Support persistent map indexes.

- **improvements**
- [CLIENT-2590] - Required Updates Following Server-Side Changes: SINDEX Support for 'Blob' Type Elements.

## [2.29.0] 2023-08-24
- **Updates**
- [CLIENT-2526] Support for set quota for user defined roles
Expand Down
34 changes: 24 additions & 10 deletions lib/aerospike/cdt/map_operation.rb
Expand Up @@ -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
Expand All @@ -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

##
Expand Down Expand Up @@ -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)

Expand All @@ -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
Expand Down
9 changes: 6 additions & 3 deletions lib/aerospike/cdt/map_policy.rb
Expand Up @@ -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
Expand All @@ -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
Expand Down
7 changes: 2 additions & 5 deletions lib/aerospike/client.rb
Expand Up @@ -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?
Expand All @@ -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)
Expand Down Expand Up @@ -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?
Expand Down
2 changes: 1 addition & 1 deletion lib/aerospike/command/batch_direct_command.rb
Expand Up @@ -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)

Expand Down
2 changes: 1 addition & 1 deletion lib/aerospike/command/batch_index_command.rb
Expand Up @@ -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)

Expand Down

0 comments on commit b9c8544

Please sign in to comment.