<commit_before>package largedatatests

import (
	"fmt"
	tc "github.com/couchbase/indexing/secondary/tests/framework/common"
	kv "github.com/couchbase/indexing/secondary/tests/framework/kvutility"
	"github.com/couchbase/indexing/secondary/tests/framework/secondaryindex"
	"log"
	"math/rand"
	"path/filepath"
	"sync"
	"testing"
	"time"
)

var kvdocs tc.KeyValues
var prodfile string

func CreateDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Creating mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		docsToBeCreated := GenerateJsons(100, seed, prodfile, bagdir)
		seed++
		kv.SetKeyValues(docsToBeCreated, "default", "", clusterconfig.KVAddress)
		for key, value := range docsToBeCreated {
			kvdocs[key] = value
		}
		// fmt.Println("CreateDocs:: Len of kvdocs", len(kvdocs))
	}
}

func DeleteDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Delete mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		i := 0
		keysToBeDeleted := make(tc.KeyValues)
		for key, value := range kvdocs {
			keysToBeDeleted[key] = value
			i++
			if i == 5 {
				break
			}
		}
		kv.DeleteKeys(keysToBeDeleted, "default", "", clusterconfig.KVAddress)
		// Update docs object with deleted keys
		for key, _ := range keysToBeDeleted {
			delete(kvdocs, key)
		}
		// fmt.Println("DeleteDocs:: Len of kvdocs", len(kvdocs))
	}
}

func CreateDeleteDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Creating mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		docsToBeCreated := GenerateJsons(100, seed, prodfile, bagdir)
		seed++
		kv.SetKeyValues(docsToBeCreated, "default", "", clusterconfig.KVAddress)
		for key, value := range docsToBeCreated {
			kvdocs[key] = value
		}
		fmt.Println("CreateDocs:: Len of kvdocs", len(kvdocs))

		i := 0
		keysToBeDeleted := make(tc.KeyValues)
		for key, value := range kvdocs {
			keysToBeDeleted[key] = value
			i++
			if i == 5 {
				break
			}
		}
		kv.DeleteKeys(keysToBeDeleted, "default", "", clusterconfig.KVAddress)
		// Update docs object with deleted keys
		for key, _ := range keysToBeDeleted {
			delete(kvdocs, key)
		}
	}
}

func RangeScanForDuration_ltr(header string, wg *sync.WaitGroup, seconds float64, t *testing.T, indexName, bucketName, server string) {
	fmt.Println("In Range Scan")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "RangeForDuration")
	defer client.Close()
	start := time.Now()
	i := 1
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		low := random_letter()
		high := random_letter()

		rangeStart := time.Now()
		scanResults, err := secondaryindex.RangeWithClient(indexName, bucketName, server, []interface{}{low}, []interface{}{high}, 3, true, defaultlimit, client)
		rangeElapsed := time.Since(rangeStart)
		fmt.Printf("Range Scan of %d user documents took %s\n", len(kvdocs), rangeElapsed)
		log.Printf("%v %d  RangeScanForDuration:: Len of scanResults is: %d\n", header, i, len(scanResults))
		i++
		FailTestIfError(err, "Error in scan", t)
	}
}

func RangeScanForDuration_num(header string, wg *sync.WaitGroup, seconds float64, t *testing.T, indexName, bucketName, server string) {
	fmt.Println("In Range Scan")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "RangeForDuration")
	defer client.Close()
	start := time.Now()
	i := 1
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		low := random_num(15, 80)
		high := random_num(15, 80)
		scanResults, err := secondaryindex.RangeWithClient(indexName, bucketName, server, []interface{}{low}, []interface{}{high}, 3, true, defaultlimit, client)
		log.Printf("%v %d  RangeScanForDuration:: Len of scanResults is: %d\n", header, i, len(scanResults))
		i++
		FailTestIfError(err, "Error in scan", t)
	}
}

func CreateDropIndexesForDuration(wg *sync.WaitGroup, seconds float64, t *testing.T) {
	fmt.Println("Create and Drop index operations")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "CDIndex")
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		var index1 = "index_age"
		var bucketName = "default"
		err := secondaryindex.CreateSecondaryIndexWithClient(index1, bucketName, indexManagementAddress, []string{"age"}, true, client)
		FailTestIfError(err, "Error in creating the index", t)
		time.Sleep(1 * time.Second)
		scanResults, err := secondaryindex.RangeWithClient(index1, bucketName, indexScanAddress, []interface{}{random_num(15, 80)}, []interface{}{random_num(15, 80)}, 3, true, defaultlimit, client)
		log.Printf("%v RangeScan CreateDropIndexesForDuration:: Len of scanResults is: %d\n", index1, len(scanResults))
		FailTestIfError(err, "Error in scan", t)

		var index2 = "index_firstname"
		err = secondaryindex.CreateSecondaryIndexWithClient(index2, bucketName, indexManagementAddress, []string{"`first-name`"}, true, client)
		FailTestIfError(err, "Error in creating the index", t)
		time.Sleep(1 * time.Second)
		scanResults, err = secondaryindex.RangeWithClient(index2, bucketName, indexScanAddress, []interface{}{"M"}, []interface{}{"Z"}, 3, true, defaultlimit, client)
		log.Printf("%v  RangeScan CreateDropIndexesForDuration:: Len of scanResults is: %d", index2, len(scanResults))
		FailTestIfError(err, "Error in scan", t)

		err = secondaryindex.DropSecondaryIndexWithClient(index1, bucketName, indexManagementAddress, client)
		FailTestIfError(err, "Error in drop index", t)
		time.Sleep(1 * time.Second)
		err = secondaryindex.DropSecondaryIndexWithClient(index2, bucketName, indexManagementAddress, client)
		FailTestIfError(err, "Error in drop index", t)
		time.Sleep(1 * time.Second)
	}
	client.Close()
}

func SequentialRangeScanForDuration(indexName, bucketName string, seconds float64, t *testing.T) {
	fmt.Println("In Range Scan")
	start := time.Now()
	i := 1<commit_msg>Added more test logging in case of scan error<commit_after>package largedatatests

import (
	"fmt"
	tc "github.com/couchbase/indexing/secondary/tests/framework/common"
	kv "github.com/couchbase/indexing/secondary/tests/framework/kvutility"
	"github.com/couchbase/indexing/secondary/tests/framework/secondaryindex"
	"log"
	"math/rand"
	"path/filepath"
	"sync"
	"testing"
	"time"
)

var kvdocs tc.KeyValues
var prodfile string

func CreateDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Creating mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		docsToBeCreated := GenerateJsons(100, seed, prodfile, bagdir)
		seed++
		kv.SetKeyValues(docsToBeCreated, "default", "", clusterconfig.KVAddress)
		for key, value := range docsToBeCreated {
			kvdocs[key] = value
		}
		// fmt.Println("CreateDocs:: Len of kvdocs", len(kvdocs))
	}
}

func DeleteDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Delete mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		i := 0
		keysToBeDeleted := make(tc.KeyValues)
		for key, value := range kvdocs {
			keysToBeDeleted[key] = value
			i++
			if i == 5 {
				break
			}
		}
		kv.DeleteKeys(keysToBeDeleted, "default", "", clusterconfig.KVAddress)
		// Update docs object with deleted keys
		for key, _ := range keysToBeDeleted {
			delete(kvdocs, key)
		}
		// fmt.Println("DeleteDocs:: Len of kvdocs", len(kvdocs))
	}
}

func CreateDeleteDocsForDuration(wg *sync.WaitGroup, seconds float64) {
	fmt.Println("CreateDocs:: Creating mutations")
	defer wg.Done()
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		docsToBeCreated := GenerateJsons(100, seed, prodfile, bagdir)
		seed++
		kv.SetKeyValues(docsToBeCreated, "default", "", clusterconfig.KVAddress)
		for key, value := range docsToBeCreated {
			kvdocs[key] = value
		}
		fmt.Println("CreateDocs:: Len of kvdocs", len(kvdocs))

		i := 0
		keysToBeDeleted := make(tc.KeyValues)
		for key, value := range kvdocs {
			keysToBeDeleted[key] = value
			i++
			if i == 5 {
				break
			}
		}
		kv.DeleteKeys(keysToBeDeleted, "default", "", clusterconfig.KVAddress)
		// Update docs object with deleted keys
		for key, _ := range keysToBeDeleted {
			delete(kvdocs, key)
		}
	}
}

func RangeScanForDuration_ltr(header string, wg *sync.WaitGroup, seconds float64, t *testing.T, indexName, bucketName, server string) {
	fmt.Println("In Range Scan")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "RangeForDuration")
	defer client.Close()
	start := time.Now()
	i := 1
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		low := random_letter()
		high := random_letter()

		rangeStart := time.Now()
		scanResults, err := secondaryindex.RangeWithClient(indexName, bucketName, server, []interface{}{low}, []interface{}{high}, 3, true, defaultlimit, client)
		rangeElapsed := time.Since(rangeStart)
		FailTestIfError(err, "RangeScanForDuration Thread:: Error in scan", t)
		fmt.Printf("Range Scan of %d user documents took %s\n", len(kvdocs), rangeElapsed)
		log.Printf("%v %d  RangeScanForDuration:: Len of scanResults is: %d\n", header, i, len(scanResults))
		i++
	}
}

func RangeScanForDuration_num(header string, wg *sync.WaitGroup, seconds float64, t *testing.T, indexName, bucketName, server string) {
	fmt.Println("In Range Scan")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "RangeForDuration")
	defer client.Close()
	start := time.Now()
	i := 1
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}
		low := random_num(15, 80)
		high := random_num(15, 80)
		scanResults, err := secondaryindex.RangeWithClient(indexName, bucketName, server, []interface{}{low}, []interface{}{high}, 3, true, defaultlimit, client)
		log.Printf("%v %d  RangeScanForDuration:: Len of scanResults is: %d\n", header, i, len(scanResults))
		i++
		FailTestIfError(err, "Error in scan", t)
	}
}

func CreateDropIndexesForDuration(wg *sync.WaitGroup, seconds float64, t *testing.T) {
	fmt.Println("Create and Drop index operations")
	defer wg.Done()
	client := secondaryindex.CreateClient(clusterconfig.KVAddress, "CDIndex")
	start := time.Now()
	for {
		elapsed := time.Since(start)
		if elapsed.Seconds() >= seconds {
			break
		}

		var index1 = "index_age"
		var bucketName = "default"
		err := secondaryindex.CreateSecondaryIndexWithClient(index1, bucketName, indexManagementAddress, []string{"age"}, true, client)
		FailTestIfError(err, "Error in creating the index", t)
		time.Sleep(1 * time.Second)
		scanResults, err := secondaryindex.RangeWithClient(index1, bucketName, indexScanAddress, []interface{}{random_num(15, 80)}, []interface{}{random_num(15, 80)}, 3, true, defaultlimit, client)
		FailTestIfError(err, "CreateDropIndexesForDuration:: Error in scan", t)
		log.Printf("%v RangeScan CreateDropIndexesForDuration:: Len of scanResults is: %d\n", index1, len(scanResults))
		
		var index2 = "index_firstname"
		err = secondaryindex.CreateSecondaryIndexWithClient(index2, bucketName, indexManagementAddress, []string{"`first-name`"}, true, client)
		FailTestIfError(err, "Error in creating the index", t)
		time.Sleep(1 * time.Second)
		scanResults, err = secondaryindex.RangeWithClient(index2, bucketName, indexScanAddress, []interface{}{"M"}, []interface{}{"Z"}, 3, true, defaultlimit, client)
		FailTestIfError(err, "CreateDropIndexesForDuration:: Error in scan", t)
		log.Printf("%v  RangeScan CreateDropIndexesForDuration:: Len of scanResults is: %d", index2, len(scanResults))

		err = secondaryindex.DropSecondaryIndexWithClient(index1, bucketName, indexManagementAddress, client)
		FailTestIfError(err, "Error in drop index", t)
		time.Sleep(1 * time.Second)
		err = secondaryindex.DropSecondaryIndexWithClient(index2, bucketName, indexManagementAddress, client)
		FailTestIfError(err, "Error in drop index", t)
		time.Sleep(1 * time.Second)
	}
	client.Close()
}

func SequentialRangeScanForDuration(indexName, bucketName string, seconds float64, t *testing.T) {
	fmt.Println("In Range Scan")
	start := time.Now()
	i := 1
{'fraction_non_alphanumeric': 0.1367766618842659, 'fraction_numerical': 0.009724214889207716, 'mean_word_length': 4.943628612032212, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2268690', 'n_tokens_mistral': 4537, 'n_tokens_neox': 4168, 'n_words': 1007}