import io.github.spencerpark.ijava.IJava;
import io.github.spencerpark.jupyter.kernel.magic.common.Shell;
IJava.getKernelInstance().getMagics().registerMagics(Shell.class);
%sh asd

%%loadFromPOM
<dependencies>
  <dependency>
    <groupId>com.aerospike</groupId>
    <artifactId>aerospike-client</artifactId>
    <version>5.0.0</version>
  </dependency>
</dependencies>

import com.aerospike.client.AerospikeClient;
import com.aerospike.client.Bin;
import com.aerospike.client.Key;
import com.aerospike.client.policy.WritePolicy;

AerospikeClient client = new AerospikeClient("localhost", 3000);
System.out.println("Initialized the client and connected to the cluster.");

String Namespace = "test";
String Set = "sql-update";

// define a convenience functions to examine all records in a namespace and set
import com.aerospike.client.Record;
import com.aerospike.client.ScanCallback;
import com.aerospike.client.policy.ScanPolicy;

public class ScanParallel implements ScanCallback {
    public void scanCallback(Key key, Record record) {
        System.out.format("namespace=%s set=%s key=%s bins=%s\n", key.namespace, key.setName, key.userKey, record.bins);
    }
}
void examineRecords(String ns, String set) {
    client.scanAll(null, ns, set, new ScanParallel());
}

// insert records in the new namespace and verify by reading back
String NamespaceCreateTest = "create-test";
String SetCreateTest = "test-set";

WritePolicy wpolicy = new WritePolicy();
wpolicy.sendKey = true;
for (int i = 1; i <= 10; i++) {
    Key key = new Key(NamespaceCreateTest, SetCreateTest, "id-"+i);
    Bin bin1 = new Bin(new String("bin1"), i);
    Bin bin2 = new Bin(new String("bin2"), 1000+i);
    client.put(wpolicy, key, bin1, bin2);
}
System.out.format("Inserted 10 records in ns=%s set=%s.\n", NamespaceCreateTest, SetCreateTest);

System.out.format("Retrieving all records in ns=%s.\n", NamespaceCreateTest);
examineRecords(NamespaceCreateTest, null); // null value for a set returns all records in the namespace

// truncate the namespace
client.truncate(null, NamespaceCreateTest, null, null); // null set value truncates all sets in the namespace

// examine records in the namespace - should be empty
System.out.format("Retrieving all records in namespace %s.\n", NamespaceCreateTest);
examineRecords(NamespaceCreateTest, null); // null set value returns all records in the namespace
System.out.format("Done.");;

import com.aerospike.client.AerospikeException;

try {
    examineRecords(NamespaceCreateTest, null); // null value for a set returns all records in the namespace
}
catch (AerospikeException ae) {
    System.out.format("Error in retrieving records from namespace %s: %s.\n", NamespaceCreateTest, ae.getMessage());
}

String SetCreateTest = "create-set";
Key key = new Key(Namespace, SetCreateTest, "id-100");
Bin bin1 = new Bin(new String("bin1"), 100);
Bin bin2 = new Bin(new String("bin2"), 1000+100);
client.put(wpolicy, key, bin1, bin2);

Key key = new Key(Namespace, SetCreateTest, "id-101");
Bin bin1 = new Bin(new String("bin1"), "string");
Bin bin3 = new Bin(new String("bin3"), "new bin");
client.put(wpolicy, key, bin1, bin3);

Key key = new Key(Namespace, SetCreateTest, "id-102");
Bin bin1 = new Bin(new String("bin1"), 111.222);
Bin bin2 = new Bin(new String("bin2"), 0.1);
Bin bin4 = new Bin(new String("bin4"), "");
client.put(wpolicy, key, bin1, bin2, bin4);

examineRecords(Namespace, SetCreateTest);

client.truncate(null, Namespace, SetCreateTest, null);

System.out.format("Retrieving all records in set=%s.\n", SetCreateTest);
examineRecords(Namespace, SetCreateTest); 
System.out.format("Done.");;

import com.aerospike.client.policy.Policy;
import com.aerospike.client.query.IndexType;
import com.aerospike.client.task.IndexTask;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.ResultCode;

String IndexName = "test_sql_update_bin1_number_idx";

Policy policy = new Policy();
policy.socketTimeout = 0; // Do not timeout on index create.

try {
    IndexTask task = client.createIndex(policy, Namespace, Set, IndexName, 
                                        "bin1", IndexType.NUMERIC);
    task.waitTillComplete();
}
catch (AerospikeException ae) {
    if (ae.getResultCode() != ResultCode.INDEX_ALREADY_EXISTS) {
        throw ae;
    }
}

System.out.format("Created number index %s on ns=%s set=%s bin=%s.", 
                                    IndexName, Namespace, Set, "bin1");;

client.dropIndex(null, Namespace, Set, IndexName);
System.out.format("Dropped index %s on ns=%s set=%s bin=%s.", 
                                    IndexName, Namespace, Set, "bin1");;

import com.aerospike.client.policy.RecordExistsAction;

Key key = new Key(Namespace, Set, "id-100");
Bin bin1 = new Bin(new String("bin1"), 100);
Bin bin2 = new Bin(new String("bin2"), 1000+100);

// ensure the record doesn't exist by removing it
try {
    client.delete(wpolicy, key);
}
catch (AerospikeException ae) {
    // ignore if doesn't exist
}

// update-only must fail since the record doesn't exist
try {
    wpolicy.recordExistsAction = RecordExistsAction.UPDATE_ONLY;
    client.put(wpolicy, key, bin1, bin2);
    System.out.format("update-only succeeded");
}
catch (AerospikeException ae) {
    System.out.format("Error on update-only: %s.\n", ae.getMessage());
}

// replace-only must fail since the record doesn't exist
try {
    wpolicy.recordExistsAction = RecordExistsAction.REPLACE_ONLY;
    client.put(wpolicy, key, bin1, bin2);
    System.out.format("replace-only succeeded");
}
catch (AerospikeException ae) {
    System.out.format("Error on replace-only: %s.\n", ae.getMessage());
}

// create-only should succeed since the record doesn't exist
try {
    wpolicy.recordExistsAction = RecordExistsAction.CREATE_ONLY;
    client.put(wpolicy, key, bin1, bin2);
    Record record = client.get(null, key);
    System.out.format("Create-only succeeded: key=%s bins=%s\n", key.userKey, record.bins);
}
catch (AerospikeException ae) {
    System.out.format("Error on create-only: %s.\n", ae.getMessage());
}

// update should succeed irrespective of the record's existence
try {
    Bin bin3 = new Bin(new String("bin3"), "new bin");
    wpolicy.recordExistsAction = RecordExistsAction.UPDATE;
    client.put(wpolicy, key, bin3);
    Record record = client.get(null, key);
    System.out.format("Update succeeded: key=%s bins=%s\n", key.userKey, record.bins);
}
catch (AerospikeException ae) {
    System.out.format("Error: %s.\n", ae.getMessage());
}

// replace should succeed irrespective of the record's existence
try {
    Bin bin4 = new Bin(new String("bin4"), "another bin");
    wpolicy.recordExistsAction = RecordExistsAction.REPLACE;
    client.put(wpolicy, key, bin4);
    Record record = client.get(null, key);
    System.out.format("Replace succeeded: key=%s bins=%s\n", key.userKey, record.bins);
}
catch (AerospikeException ae) {
    System.out.format("Error: %s.\n", ae.getMessage());
}

Key key = new Key(Namespace, Set, "id-100");
Bin bin1 = new Bin(new String("bin1"), 100);
Bin bin2 = new Bin(new String("bin2"), "John");

wpolicy.recordExistsAction = RecordExistsAction.REPLACE;
client.put(wpolicy, key, bin1, bin2);
Record record = client.get(null, key);
System.out.format("Record: key=%s bins=%s\n", key.userKey, record.bins);

// add
bin1 = new Bin(new String("bin1"), 1);
client.add(null, key, bin1);
Record record = client.get(null, key);
System.out.format("After add: key=%s bins=%s\n", key.userKey, record.bins);

// append
bin2 = new Bin(new String("bin2"), " Doe");
client.append(null, key, bin2);
Record record = client.get(null, key);
System.out.format("After append: key=%s bins=%s\n", key.userKey, record.bins);

//prepend
bin2 = new Bin(new String("bin2"), "Mr ");
client.prepend(null, key, bin2);
Record record = client.get(null, key);
System.out.format("After prepend: key=%s bins=%s\n", key.userKey, record.bins);;

// create a record
Key key = new Key(Namespace, Set, "id-100");
Bin bin1 = new Bin(new String("bin1"), 100);
Bin bin2 = new Bin(new String("bin2"), 1000+100);
client.put(wpolicy, key, bin1, bin2);
Record record = client.get(null, key);
System.out.format("Record: key=%s bins=%s\n", key.userKey, record.bins);

// delete the record
client.delete(wpolicy, key);
System.out.format("Record deleted: key=%s\n", key.userKey);

// verify it's deleted
Record record = client.get(null, key);
if (record == null) {
    System.out.format("No record found for key=%s \n", key.userKey);;
} 
else {
    System.out.format("Record: key=%s bins=%s\n", key.userKey, record.bins);;
}

import com.aerospike.client.Operation;

// create a record 
Key key = new Key(Namespace, Set, "id-100");
Bin bin1 = new Bin(new String("bin1"), 100);
Bin bin2 = new Bin(new String("bin2"), "John");
client.put(wpolicy, key, bin1, bin2);
Record record = client.get(null, key);
System.out.format("Record: key=%s bins=%s\n", key.userKey, record.bins);

// Add integer, update string, read record in one operate call
bin1 = new Bin(new String("bin1"), 1);
bin2 = new Bin(new String("bin2"), " Doe");
record = client.operate(null, key, Operation.add(bin1), Operation.append(bin2), Operation.get());
System.out.format("Operate results: key=%s bins=%s\n", key.userKey, record.bins);;

// populate data, create secondary index, and define a convenience function

// add records with keys "id-1" to "id-10" and 
//    bins bin1 (integer values 1-10) and bin2 (integer values 1000-1010).
WritePolicy wpolicy = new WritePolicy();
wpolicy.sendKey = true;
for (int i = 1; i <= 10; i++) {
    Key key = new Key(Namespace, Set, "id-"+i);
    Bin bin1 = new Bin(new String("bin1"), i);
    Bin bin2 = new Bin(new String("bin2"), 1000+i);
    client.put(wpolicy, key, bin1, bin2);
}
System.out.format("Test data popuated.\n");;

// create an integer secondary index on bin1
import com.aerospike.client.policy.Policy;
import com.aerospike.client.query.IndexType;
import com.aerospike.client.task.IndexTask;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.ResultCode;

String IndexName = "test_sql_update_bin1_number_idx";

Policy policy = new Policy();
policy.socketTimeout = 0; // Do not timeout on index create.
try {
    IndexTask task = client.createIndex(policy, Namespace, Set, IndexName, 
                                        "bin1", IndexType.NUMERIC);
    task.waitTillComplete();
}
catch (AerospikeException ae) {
    if (ae.getResultCode() != ResultCode.INDEX_ALREADY_EXISTS) {
        throw ae;
    }
}
System.out.format("Created number index %s on ns=%s set=%s bin=%s.\n", 
                                    IndexName, Namespace, Set, "bin1");;
                                    
// define convenience functions
import com.aerospike.client.query.Statement;
import com.aerospike.client.query.Filter;
import com.aerospike.client.Record;

void printRecordRange(String header, String namespace, String set, String bin, int rangeLow, int rangeHigh) {
    System.out.format("%s\n", header);
    Statement stmt = new Statement();
    stmt.setNamespace(namespace);
    stmt.setSetName(set);
    stmt.setFilter(Filter.range(bin, rangeLow, rangeHigh));
    RecordSet rs = client.query(null, stmt);
    while (rs.next()) {
        Key key = rs.getKey();
        Record record = rs.getRecord();
        System.out.format("key=%s bins=%s\n", key.userKey, record.bins);
    }
    rs.close();
}

import com.aerospike.client.Value;
import com.aerospike.client.query.RecordSet;
import com.aerospike.client.policy.QueryPolicy;
import com.aerospike.client.exp.Exp;
import com.aerospike.client.Operation;
import com.aerospike.client.task.ExecuteTask;

// 1. list of updates specified in Statement 
printRecordRange("initial state:", Namespace, Set, "bin1", 4, 7);

Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
stmt.setFilter(Filter.range("bin1", 4, 7));
Operation ops[] = { new Operation(Operation.Type.ADD, "bin2", Value.get(1)), 
                    Operation.put(new Bin("bin3", "new1"))};
stmt.setOperations(ops);
ExecuteTask task = client.execute(null, stmt); 
task.waitTillComplete(3000, 3000);

printRecordRange("After executing updates in Statement with filter:", Namespace, Set, "bin1", 4, 7);


// 2. list of updates as argument to execute

printRecordRange("initial state:", Namespace, Set, "bin1", 4, 7);

Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
stmt.setFilter(Filter.range("bin1", 4, 7));

ExecuteTask task = client.execute(null, stmt, new Operation(Operation.Type.ADD, "bin2", Value.get(1)), 
                                            Operation.put(new Bin("bin3", "new2"))); 
task.waitTillComplete(3000, 3000);

printRecordRange("After executing updates in Ops list argument:", Namespace, Set, "bin1", 4, 7);


// 3. updates in Statement as well as in the execute argument - Statement operations are ignored

printRecordRange("initial state:", Namespace, Set, "bin1", 4, 7);

Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
stmt.setFilter(Filter.range("bin1", 4, 7));
Operation ops[] = { new Operation(Operation.Type.ADD, "bin2", Value.get(1))};
stmt.setOperations(ops);
WritePolicy policy = new WritePolicy();
policy.maxRetries = 0;

ExecuteTask task = client.execute(policy, stmt, Operation.put(new Bin("bin3", "new3"))); 
task.waitTillComplete(3000, 3000);

printRecordRange("After executing updates in Statement and Ops list argument (Statement updates are ignored):", 
                    Namespace, Set, "bin1", 4, 7);

import com.aerospike.client.policy.Policy;
import com.aerospike.client.task.RegisterTask;
import com.aerospike.client.Language;
import com.aerospike.client.lua.LuaConfig;
import com.aerospike.client.lua.LuaCache;

LuaConfig.SourceDirectory = "../udf";
String UDFFile = "update_example.lua";
String UDFModule = "update_example";

void registerUDF() {
    // clear the lua cache
    LuaCache.clearPackages();
    Policy policy = new Policy();
    // remove the current module, if any
    client.removeUdf(null, UDFFile);
    RegisterTask task = client.register(policy, LuaConfig.SourceDirectory+"/"+UDFFile, 
                                        UDFFile, Language.LUA);
    task.waitTillComplete();
    System.out.format("Registered the UDF module %s.", UDFFile);;
}

registerUDF();

printRecordRange("initial state:", Namespace, Set, "bin1", 4, 7);

Statement stmt = new Statement();
stmt.setNamespace(Namespace);
stmt.setSetName(Set);
WritePolicy policy = new WritePolicy();
policy.filterExp = Exp.build(
    Exp.and(
        Exp.ge(Exp.intBin("bin1"), Exp.val(4)),
        Exp.le(Exp.intBin("bin1"), Exp.val(7))));
policy.maxRetries = 0;

ExecuteTask task = client.execute(policy, stmt, UDFModule, "add_append",
                                 Value.get("bin2"), Value.get(1), Value.get("bin3"), Value.get("!!!")); 
task.waitTillComplete(3000, 3000);

printRecordRange("After executing UDF add_append:", 
                    Namespace, Set, "bin1", 4, 7);

client.dropIndex(null, Namespace, Set, IndexName);
client.truncate(null, Namespace, null, null);
client.close();
System.out.println("Removed tutorial data and closed server connection.");