CLIENT-2706 In scan/query with max_records set, count discarded records that would have exceeded max_records. This count is used to correctly determine if a node’s partitions should be retried on the next scan/query page.

Author: BrianNichols

src/include/aerospike/as_partition_tracker.h

@@ -44,6 +44,7 @@ typedef struct as_node_partitions_s {
 	as_vector parts_partial;
 	uint64_t record_count;
 	uint64_t record_max;
+	uint64_t disallowed_count;
 	uint32_t parts_unavailable;
 } as_node_partitions;
 
@@ -138,17 +139,29 @@ as_partition_tracker_set_last(
 }
 
 static inline bool
-as_partition_tracker_reached_max_records_sync(as_partition_tracker* pt)
+as_partition_tracker_reached_max_records_sync(as_partition_tracker* pt, as_node_partitions* np)
 {
 	// Sync scan/query runs in multiple threads, so atomics are required.
-	return pt && pt->check_max && (as_aaf_uint64(&pt->record_count, 1) > pt->max_records);
+	if (pt && pt->check_max && (as_aaf_uint64(&pt->record_count, 1) > pt->max_records)) {
+		// Record was returned, but would exceed max_records.
+		// Discard record and increment disallowed_count.
+		np->disallowed_count++;
+		return true;
+	}
+	return false;
 }
 
 static inline bool
-as_partition_tracker_reached_max_records_async(as_partition_tracker* pt)
+as_partition_tracker_reached_max_records_async(as_partition_tracker* pt, as_node_partitions* np)
 {
 	// Async scan/query runs in a single event loop thread, so atomics are not necessary.
-	return pt && pt->check_max && (++pt->record_count > pt->max_records);
+	if (pt && pt->check_max && (++pt->record_count > pt->max_records)) {
+		// Record was returned, but would exceed max_records.
+		// Discard record and increment disallowed_count.
+		np->disallowed_count++;
+		return true;
+	}
+	return false;
 }
 
 static inline uint16_t

src/main/aerospike/aerospike_query.c

@@ -336,7 +336,7 @@ as_query_parse_record_async(
 		return status;
 	}
 
-	if (as_partition_tracker_reached_max_records_async(qe->pt)) {
+	if (as_partition_tracker_reached_max_records_async(qe->pt, qc->np)) {
 		as_record_destroy(&rec);
 		return AEROSPIKE_OK;
 	}
@@ -471,7 +471,7 @@ as_query_parse_record(uint8_t** pp, as_msg* msg, as_query_task* task, as_error*
 			return status;
 		}
 
-		if (as_partition_tracker_reached_max_records_sync(task->pt)) {
+		if (as_partition_tracker_reached_max_records_sync(task->pt, task->np)) {
 			as_record_destroy(&rec);
 			return AEROSPIKE_OK;
 		}

src/main/aerospike/aerospike_scan.c

@@ -201,7 +201,7 @@ as_scan_parse_record_async(
 		return status;
 	}
 
-	if (as_partition_tracker_reached_max_records_async(se->pt)) {
+	if (as_partition_tracker_reached_max_records_async(se->pt, sc->np)) {
 		as_record_destroy(&rec);
 		return AEROSPIKE_OK;
 	}
@@ -305,7 +305,7 @@ as_scan_parse_record(uint8_t** pp, as_msg* msg, as_scan_task* task, as_error* er
 		return status;
 	}
 
-	if (as_partition_tracker_reached_max_records_sync(task->pt)) {
+	if (as_partition_tracker_reached_max_records_sync(task->pt, task->np)) {
 		as_record_destroy(&rec);
 		return AEROSPIKE_OK;
 	}

src/main/aerospike/as_partition_tracker.c

@@ -446,7 +446,7 @@ as_partition_tracker_is_complete(as_partition_tracker* pt, as_cluster* cluster,
 				for (uint32_t i = 0; i < list->size; i++) {
 					as_node_partitions* np = as_vector_get(list, i);
 
-					if (np->record_count >= np->record_max) {
+					if (np->record_count + np->disallowed_count >= np->record_max) {
 						mark_retry(pt, np);
 						is_done = false;
 					}
@@ -460,7 +460,7 @@ as_partition_tracker_is_complete(as_partition_tracker* pt, as_cluster* cluster,
 				for (uint32_t i = 0; i < list->size; i++) {
 					as_node_partitions* np = as_vector_get(list, i);
 
-					if (np->record_count > 0) {
+					if (np->record_count + np->disallowed_count > 0) {
 						mark_retry(pt, np);
 					}
 				}