{ "cells": [ { "cell_type": "markdown", "metadata": { "toc": true }, "source": [ "

Table of Contents

\n", "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Implementing SQL Operations: SELECT\n", "This tutorial describes how to implement certain SQL SELECT statements in Aerospike.\n", "\n", "This notebook requires the Aerospike Database running locally with Java kernel and Aerospike Java Client. To create a Docker container that satisfies the requirements and holds a copy of Aerospike notebooks, visit the [Aerospike Notebooks Repo](https://github.com/aerospike-examples/interactive-notebooks)." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Introduction\n", "In this notebook, we will see how specific SELECT statements in SQL can be implemented in Aerospike. \n", "\n", "SQL is a widely known data access language. If you have used SQL, the examples in this notebook will make it easier to implement specific SQL SELECT statements. \n", "\n", "This notebook is the first in the SQL Operations series that consists of the following notebooks:\n", "- Implementing SQL Operations: SELECT (this notebook)\n", "- Implementing SQL Operations: UPDATE, CREATE, and DELETE\n", "- Implementing SQL Operations: Aggregates\n", "\n", "The specific topics and SQL SELECT statements we discuss include:\n", "- Components of the SELECT statement\n", "- Single record retrieval\n", "- Batch retrieval\n", "- Predicate based retrieval\n", " - Query operation using an index\n", " - Query operation using an expression filter\n", " - Scan operation using an expression filter\n", "- Computed fields\n", "\n", "The purpose of this notebook is to provide Aerospike API equivalents for specific SQL operations. Not all SQL operations (such as JOIN) are directly supported in Aerospike API. The SQL syntax in this notebook while largely accurate is meant only to convey the semantics of the operations. Check out [Aerospike Presto Connector](https://www.aerospike.com/docs/connect/access/presto/index.html) for ad-hoc SQL access to Aerospike data.\n", "\n", "While Aerospike provides both synchronous and asynchronous execution modes for many operations, we use mostly synchronous execution examples in this notebook, leaving asynchronous execution as a separate topic for a future tutorial. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Prerequisites\n", "This tutorial assumes familiarity with the following topics:\n", "- [Hello World](hello_world.ipynb)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Initialization" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Ensure database is running\n", "This notebook requires that Aerospike database is running. " ] }, { "cell_type": "code", "execution_count": 53, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:49.065421Z", "start_time": "2020-12-29T20:48:49.060897Z" } }, "outputs": [], "source": [ "import io.github.spencerpark.ijava.IJava;\n", "import io.github.spencerpark.jupyter.kernel.magic.common.Shell;\n", "IJava.getKernelInstance().getMagics().registerMagics(Shell.class);\n", "%sh asd" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Download and install additional components.\n", "Install the Java client." ] }, { "cell_type": "code", "execution_count": 54, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:50.084636Z", "start_time": "2020-12-29T20:48:50.080629Z" } }, "outputs": [], "source": [ "%%loadFromPOM\n", "\n", " \n", " com.aerospike\n", " aerospike-client\n", " 5.0.0\n", " \n", "" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Connect to database and populate test data\n", "The test data has ten records with user-key \"id-1\" through \"id-10\", two integer bins (fields) \"bin1\" and \"bin2\", in the namespace \"test\" and sets \"sql-select-small\"and null, and similarly structured 1000 records in set \"sql-select-large\". " ] }, { "cell_type": "code", "execution_count": 55, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:50.771243Z", "start_time": "2020-12-29T20:48:50.767819Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Initialized the client and connected to the cluster.\n", "Test data populated" ] } ], "source": [ "import com.aerospike.client.AerospikeClient;\n", "import com.aerospike.client.Bin;\n", "import com.aerospike.client.Key;\n", "import com.aerospike.client.policy.WritePolicy;\n", "\n", "AerospikeClient client = new AerospikeClient(\"localhost\", 3000);\n", "System.out.println(\"Initialized the client and connected to the cluster.\");\n", "\n", "String Namespace = \"test\";\n", "String SmallSet = \"sql-select-small\";\n", "String LargeSet = \"sql-select-large\";\n", "String NullSet = \"\";\n", "\n", "WritePolicy wpolicy = new WritePolicy();\n", "wpolicy.sendKey = true;\n", "for (int i = 1; i <= 10; i++) {\n", " Key key = new Key(Namespace, SmallSet, \"id-\"+i);\n", " Bin bin1 = new Bin(new String(\"bin1\"), i);\n", " Bin bin2 = new Bin(new String(\"bin2\"), 1000+i);\n", " client.put(wpolicy, key, bin1, bin2);\n", "}\n", "for (int i = 1; i <= 10; i++) {\n", " Key key = new Key(Namespace, NullSet, \"id-\"+i);\n", " Bin bin1 = new Bin(new String(\"bin1\"), i);\n", " Bin bin2 = new Bin(new String(\"bin2\"), 1000+i);\n", " client.put(wpolicy, key, bin1, bin2);\n", "}\n", "for (int i = 1; i <= 1000; i++) {\n", " Key key = new Key(Namespace, LargeSet, \"id-\"+i);\n", " Bin bin1 = new Bin(new String(\"bin1\"), i);\n", " Bin bin2 = new Bin(new String(\"bin2\"), 1000+i);\n", " client.put(wpolicy, key, bin1, bin2);\n", "}\n", "\n", "System.out.format(\"Test data populated\");;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Mapping Components of SELECT Statement\n", "**Columns and tables**\n", "\n", "In Aerospike, a relational database or schema maps to a namespace, a table maps to a set, and a column maps to a bin. Thus a query `SELECT columns FROM table WHERE condition` can be written in Aerospike terminology as `SELECT bins FROM namespace.set WHERE condition`.\n", "\n", "**Record id**\n", "\n", "Records are stored in a namespace, organized in sets, and each record is uniquely identified by a key or id of the record that consists of a triple: (namespace, set, user-key) where user-key is a unique user specified id within the set. The key is closely identified with a record, and can be seen either as a metadata or a primary key field, and is returned in all retrieval APIs.\n", "\n", "**Record metadata**\n", "\n", "Each record has generation (or version) and expiration (or time-to-live in seconds) associated with it. This metadata is returned in all retrieval operations. It is possible to retrieve only the metadata without the record's bins through \"getHeader\" operation explained below. \n", "\n", "**A note on Policy**\n", "\n", "All APIs take a Policy argument. A policy specifies many request parameters such as timeout and maximum retries, as well as operations modifiers such as an expression filter. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Single Record Get\n", "Let's start with a simple example of a single record retrieval using its key. You can either get the entire record or specific bins.\n", "\n", "`SELECT * FROM namespace.set WHERE id = key`\n", "- `Record Client::get(Policy policy, Key key)`\n", "\n", "`SELECT bins FROM namsepace.set WHERE id = key`\n", "- `Record Client::get(Policy policy, Key key, String... binNames)`" ] }, { "cell_type": "code", "execution_count": 56, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "set='sql-select-small' key=id-3 bins={bin1=3, bin2=1003}\n", "set='' key=id-4 bins={bin2=1004}\n" ] } ], "source": [ "import com.aerospike.client.Record;\n", "\n", "// Read all bins of the record with user-key \"id-3\" from the small set\n", "Key key = new Key(Namespace, SmallSet, \"id-3\");\n", "// retrieve all bins in the record\n", "Record record = client.get(null, key);\n", "System.out.format(\"set='%s' key=%s bins=%s\\n\", key.setName, key.userKey, record.bins);\n", "\n", "// Read \"bin2\" for the record with user-key \"id-4\" from the null set\n", "Key key = new Key(Namespace, NullSet, \"id-4\");\n", "// retrieve only \"bin2\"\n", "Record record = client.get(null, key, \"bin2\");\n", "System.out.format(\"set='%s' key=%s bins=%s\\n\", key.setName, key.userKey, record.bins);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Single Record Existence\n", "There is a variant of single record retrieval to check a record's existence. \n", "\n", "`SELECT EXISTS(SELECT * FROM namespace.set WHERE id = key)`\n", "- `boolean Client::exists(Policy policy, Key key)`" ] }, { "cell_type": "code", "execution_count": 57, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "key=id-3 exists=true\n", "key=no-such-key exists=false\n" ] } ], "source": [ "// Check existence of the record with user-key \"id-3\" in the small set\n", "Key key = new Key(Namespace, SmallSet, \"id-3\");\n", "boolean yesNo = client.exists(null, key);\n", "System.out.format(\"key=%s exists=%s\\n\", key.userKey, yesNo);\n", "\n", "// Check existence of the record with user-key \"no-such-key\" in the small set\n", "Key key = new Key(Namespace, SmallSet, \"no-such-key\");\n", "boolean yesNo = client.exists(null, key);\n", "System.out.format(\"key=%s exists=%s\\n\", key.userKey, yesNo);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Single Record Metadata\n", "It is possible to only obtain a record's header info or metadata, consisting of generation (or version) and expiration time (time-to-live in seconds).\n", "\n", "`SELECT generation, expiration FROM namespace.set WHERE id = key`\n", "- `Record Client::getHeader(Policy policy, Key key)`" ] }, { "cell_type": "code", "execution_count": 58, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "key=id-3 generation=1 expiration=355535130\n" ] } ], "source": [ "// Get the metadata for the record with user-key \"id-3\" in the small set\n", "Key key = new Key(Namespace, SmallSet, \"id-3\");\n", "Record header = client.getHeader(null, key);\n", "System.out.format(\"key=%s generation=%s expiration=%s\\n\", key.userKey, header.generation, header.expiration);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Batch Retrieval\n", "A batch operation operates on a list of records identified by the keys provided. This works similar to a single record retrieval, except multiple records are returned.\n", "\n", "`SELECT * FROM namespace.set WHERE id IN key-list`\n", "- `Record[] Client::get(BatchPolicy policy, Key[] keys)`\n", "\n", "`SELECT bins FROM namespace.set WHERE id in key-list`\n", "- `Record[] Client::get(BatchPolicy policy, Key[] keys, String... binNames)`" ] }, { "cell_type": "code", "execution_count": 59, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Batch retrieval results:\n", "Key not found: key=id-0\n", "key=id-1 bins={bin1=1, bin2=1001}\n", "key=id-2 bins={bin1=2, bin2=1002}\n", "key=id-3 bins={bin1=3, bin2=1003}\n" ] } ], "source": [ "// Records with user-keys \"id-0\" through \"id-3\" from the small set\n", "Key[] keys = new Key[4];\n", "for (int i = 0; i < 4; i++) {\n", " keys[i] = new Key(Namespace, SmallSet, \"id-\"+i);\n", "}\n", "// Get \"bin1\" and \"bin2\"\n", "Record[] records = client.get(null, keys, \"bin1\", \"bin2\");\n", "System.out.format(\"Batch retrieval results:\\n\");\n", "for (int i = 0; i < records.length; i++) {\n", " Key key = keys[i];\n", " Record record = records[i];\n", " if (record != null) {\n", " System.out.format(\"key=%s bins=%s\\n\", key.userKey, record.bins);\n", " }\n", " else {\n", " System.out.format(\"Key not found: key=%s\\n\", key.userKey); \n", " }\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Record Existence\n", "There is a variant of batch retrieval to check record existence. \n", "\n", "`SELECT id, EXISTS(SELECT * FROM namespace.set WHERE id = key) WHERE key IN key-list`\n", "- `boolean[] Client::exists(Policy policy, Key[] keys)`" ] }, { "cell_type": "code", "execution_count": 60, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Batch exists results:\n", "key=id-0 exists=false\n", "key=id-1 exists=true\n", "key=id-2 exists=true\n", "key=id-3 exists=true\n" ] } ], "source": [ "// Check existence of records with user-keys \"id-0\" through \"id-3\" in the small set\n", "Key[] keys = new Key[4];\n", "for (int i = 0; i < 4; i++) {\n", " keys[i] = new Key(Namespace, SmallSet, \"id-\"+i);\n", "}\n", "boolean yesNo[] = client.exists(null, keys);\n", "System.out.format(\"Batch exists results:\\n\");\n", "for (int i = 0; i < 4; i++) {\n", " System.out.format(\"key=%s exists=%s\\n\", keys[i].userKey, yesNo[i]);\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Record Metadata\n", "It is possible to obtain header info or metadata consisting of generation (or version) and expiration time (time-to-live in seconds) for a specified set of records.\n", "\n", "`SELECT generation, expiration FROM namespace.set WHERE id IN key-list`\n", "- `Record[] Client::getHeader(Policy policy, Key[] keys)`" ] }, { "cell_type": "code", "execution_count": 61, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Batch metadata results:\n", "Key not found: key=id-0\n", "key=id-1 generation=1 expiration=355535130\n", "key=id-2 generation=1 expiration=355535130\n", "key=id-3 generation=1 expiration=355535130\n" ] } ], "source": [ "// Header info for records with user-keys \"id-0\" through \"id-3\" in the small set\n", "Key[] keys = new Key[4];\n", "for (int i = 0; i < 4; i++) {\n", " keys[i] = new Key(Namespace, SmallSet, \"id-\"+i);\n", "}\n", "Record headers[] = client.getHeader(null, keys);\n", "System.out.format(\"Batch metadata results:\\n\");\n", "for (int i = 0; i < 4; i++) {\n", " if (headers[i] != null) {\n", " System.out.format(\"key=%s generation=%s expiration=%s\\n\", keys[i].userKey, \n", " headers[i].generation, headers[i].expiration);\n", " }\n", " else {\n", " System.out.format(\"Key not found: key=%s\\n\", keys[i].userKey); \n", " }\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Union of Batch Retrievals\n", "A more general form of batch reads is also available that provides a union of simple batch results with different namespace, set, and bin specification. It populates the argument \"records\" on return.\n", "\n", "`(SELECT bins1 FROM namespace1.set1 WHERE id IN key-list1) UNION (SELECT bins2 FROM namespace2.set2 WHERE id IN key-list2) UNION ...`\n", "- `void Client::get(BatchPolicy policy, List records)`\n" ] }, { "cell_type": "code", "execution_count": 62, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Union of multi batch results:\n", "set='sql-select-small' key=id-1 bins={bin2=1001}\n", "set='' key=id-1 bins={bin1=1, bin2=1001}\n", "set='sql-select-small' key=id-2 bins=null\n", "set='sql-select-small' key=id-3 bins=null\n", "Key not found: set='sql-select-small' key=no-such-key\n" ] } ], "source": [ "import java.util.List;\n", "import com.aerospike.client.BatchRead;\n", "\n", "// Array of records to specify batch parameters and get back results\n", "List records = new ArrayList();\n", "\n", "// bin2 of record id-1 from the small set\n", "records.add(new BatchRead(new Key(Namespace, SmallSet, \"id-\" + 1), new String[]{\"bin2\"}));\n", "\n", "// All bins of record id-1 from the null set\n", "records.add(new BatchRead(new Key(Namespace, NullSet, \"id-\" + 1), true));\n", "\n", "// No data bins (only headers) of record id-2 from the small set\n", "records.add(new BatchRead(new Key(Namespace, SmallSet, \"id-\" + 2), false));\n", "\n", "// This record should be found, but the requested bin will not be found.\n", "records.add(new BatchRead(new Key(Namespace, SmallSet, \"id-\" + 3), new String[]{\"no-such-bin\"}));\n", "\n", "// This record should not be found.\n", "records.add(new BatchRead(new Key(Namespace, SmallSet, \"no-such-key\"), true));\n", "\n", "// Execute batch. Note, records array is populated on return.\n", "client.get(null, records);\n", "\n", "System.out.format(\"Union of multi batch results:\\n\");\n", "for (BatchRead record : records) {\n", " Key key = record.key;\n", " Record rec = record.record;\n", " if (rec != null) {\n", " System.out.format(\"set='%s' key=%s bins=%s\\n\", key.setName, key.userKey, rec.bins);\n", " }\n", " else {\n", " System.out.format(\"Key not found: set='%s' key=%s\\n\", key.setName, key.userKey);\n", " }\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Predicate Based Retrieval\n", "In these operations, records matching a general predicate (or a condition) are retrieved.\n", "\n", "`SELECT bins FROM namespace.set WHERE condition`\n", "\n", "There are multiple ways of performing this SQL query in Aerospike. They involve query and scan operations. \n", "- Query operation using an index and/or expression filter\n", "- Scan operation using an expression filter\n", "\n", "The query operation must be used when an index is involved, but may be used without an index. The scan operation can only be used without an index. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Query Based on Index\n", "In SQL, an index if applicable is used automatically. In Aerospike, one must know the index and specify it explicitly in the statement argument in a query operation.\n", "\n", "`Record[] query(QueryPolicy policy, Statement statement)`" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Create a secondary index\n", "To use the query API with index based filter, a secondary index must exist on the filter bin. Here we create a numeric index on \"bin1\" in \"sql-select-small\" set." ] }, { "cell_type": "code", "execution_count": 63, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:48:53.066005Z", "start_time": "2020-12-29T20:48:53.061118Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Created index test_small_bin1_number_idx on ns=test set=sql-select-small bin=bin1." ] } ], "source": [ "import com.aerospike.client.policy.Policy;\n", "import com.aerospike.client.query.IndexType;\n", "import com.aerospike.client.task.IndexTask;\n", "import com.aerospike.client.AerospikeException;\n", "import com.aerospike.client.ResultCode;\n", "\n", "String IndexName = \"test_small_bin1_number_idx\";\n", "\n", "Policy policy = new Policy();\n", "policy.socketTimeout = 0; // Do not timeout on index create.\n", "\n", "try {\n", " IndexTask task = client.createIndex(policy, Namespace, SmallSet, IndexName, \n", " \"bin1\", IndexType.NUMERIC);\n", " task.waitTillComplete();\n", "}\n", "catch (AerospikeException ae) {\n", " if (ae.getResultCode() != ResultCode.INDEX_ALREADY_EXISTS) {\n", " throw ae;\n", " }\n", "}\n", "\n", "System.out.format(\"Created index %s on ns=%s set=%s bin=%s.\", \n", " IndexName, Namespace, SmallSet, \"bin1\");;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Run the Query Based on Index " ] }, { "cell_type": "code", "execution_count": 64, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Query based on bin1 index results\n", "key=id-1 bins={bin1=1, bin2=1001}\n", "key=id-2 bins={bin1=2, bin2=1002}\n", "key=id-3 bins={bin1=3, bin2=1003}\n" ] } ], "source": [ "import com.aerospike.client.query.Statement;\n", "import com.aerospike.client.query.Filter;\n", "import com.aerospike.client.Value;\n", "import com.aerospike.client.query.RecordSet;\n", "import com.aerospike.client.Record;\n", "import com.aerospike.client.policy.QueryPolicy;\n", "import com.aerospike.client.exp.Exp;\n", "\n", "Statement stmt = new Statement();\n", "stmt.setNamespace(Namespace);\n", "stmt.setSetName(SmallSet);\n", "stmt.setFilter(Filter.range(\"bin1\", 1, 3));\n", "\n", "RecordSet rs = client.query(null, stmt);\n", "\n", "System.out.format(\"Query based on bin1 index results\\n\");\n", "while (rs.next()) {\n", " Key key = rs.getKey();\n", " Record record = rs.getRecord();\n", " System.out.format(\"key=%s bins=%s\\n\", key.userKey, record.bins);\n", "}\n", "rs.close();" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Query Based on Expression Filter\n", "A general condition is specified as an expression filter, which does not use any underlying index. A synchronous call returns an array of records.\n", "\n", "`Record[] query(QueryPolicy policy, Statement statement)`" ] }, { "cell_type": "code", "execution_count": 65, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Query based on expression results\n", "key=id-2 bins={bin1=2, bin2=1002}\n", "key=id-3 bins={bin1=3, bin2=1003}\n", "key=id-1 bins={bin1=1, bin2=1001}\n" ] } ], "source": [ "Statement stmt = new Statement();\n", "stmt.setNamespace(Namespace);\n", "stmt.setSetName(SmallSet);\n", "\n", "// expression filter 1001 <= bin2 <= 1003 is specified in the operation policy\n", "QueryPolicy policy = new QueryPolicy(client.queryPolicyDefault);\n", "policy.filterExp = Exp.build(\n", " Exp.and(\n", " Exp.ge(Exp.intBin(\"bin2\"), Exp.val(1001)),\n", " Exp.le(Exp.intBin(\"bin2\"), Exp.val(1003))));\n", "\n", "RecordSet rs = client.query(policy, stmt);\n", "\n", "System.out.format(\"Query based on expression results\\n\");\n", "while (rs.next()) {\n", " Key key = rs.getKey();\n", " Record record = rs.getRecord();\n", " System.out.format(\"key=%s bins=%s\\n\", key.userKey, record.bins);\n", "}\n", "rs.close();" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Scan Based on Expression Filter\n", "The scan operation takes a callback object which is called for every record in the result.\n", "\n", "`void scanAll(ScanPolicy policy, String namespace, String setName, ScanCallback callback, String... binNames)`" ] }, { "cell_type": "code", "execution_count": 66, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Scan with expression filter results:\n", "key=id-2 bins={bin1=2, bin2=1002}\n", "key=id-3 bins={bin1=3, bin2=1003}\n", "key=id-1 bins={bin1=1, bin2=1001}\n" ] } ], "source": [ "import com.aerospike.client.ScanCallback;\n", "import com.aerospike.client.policy.ScanPolicy;\n", "\n", "public class ScanParallel implements ScanCallback {\n", "\tpublic void scanCallback(Key key, Record record) {\n", " System.out.format(\"key=%s bins=%s\\n\", key.userKey, record.bins);\n", " }\n", "}\n", "// expression filter 1 <= bin1 <= 3 is specified in the operation policy\n", "ScanPolicy policy = new ScanPolicy();\n", "policy.filterExp = Exp.build(\n", " Exp.and(\n", " Exp.ge(Exp.intBin(\"bin1\"), Exp.val(1)),\n", " Exp.le(Exp.intBin(\"bin1\"), Exp.val(3))));\n", "System.out.format(\"Scan with expression filter results:\\n\");\n", "client.scanAll(policy, Namespace, SmallSet, new ScanParallel());" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Key Points\n", "Here are some key points to remember about query and scan operations in Aerospike.\n", "- To leverage an index, one must use a query operation.\n", "- A query takes either or both: an index predicate and an expression filter.\n", "- An expression filter may be used instead of an index predicate, but it will not perform as well.\n", "- When only an expression filter is needed, either a query or a scan may be used (as shown above).\n", "- A null set value when an index predicate is used works on the null set (records belonging to no set), but without an index predicate works on the entire namespace.\n", "- An expression filter is specified within the policy, and is applied generally for filtering records beyond query and scan. You can find examples of this outside of this tutorial.\n", "\n", "Some of these are illustrated with examples in the following cells." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### To leverage an index, one must use a query operation.\n", "If an index predicate is used on an unindexed bin, it results in an error." ] }, { "cell_type": "code", "execution_count": 67, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Query with index predicate on unindexed bin results:\n", "Error: com.aerospike.client.AerospikeException: Error 201,1,30000,0,5,BB9020011AC4202 127.0.0.1 3000: Index not found" ] } ], "source": [ "Statement stmt = new Statement();\n", "stmt.setNamespace(Namespace);\n", "stmt.setSetName(SmallSet);\n", "\n", "// try to use an index predicate on bin2 which has no index\n", "stmt.setFilter(Filter.range(\"bin2\", 1004, 1007));\n", "\n", "try {\n", " RecordSet rs = client.query(null, stmt);\n", "\n", " System.out.format(\"Query with index predicate on unindexed bin results:\\n\");\n", " while (rs.next()) {\n", " Key key = rs.getKey();\n", " Record record = rs.getRecord();\n", " System.out.format(\"key=%s bin1=%s bin2=%s\\n\", key.userKey, record.bins);\n", " }\n", " rs.close();\n", "}\n", "catch (AerospikeException ae) {\n", " System.out.format(\"Error: %s\", ae);\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### A query takes either or both: an index predicate and an expression filter.\n", "Below is an example of using both of them in a query. When both are specified, the two are ANDed." ] }, { "cell_type": "code", "execution_count": 68, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Query based on index predicate and expression results\n", "key=id-2 bins={bin1=2, bin2=1002}\n", "key=id-3 bins={bin1=3, bin2=1003}\n" ] } ], "source": [ "Statement stmt = new Statement();\n", "stmt.setNamespace(Namespace);\n", "stmt.setSetName(SmallSet);\n", "// index predicate to get bin1 in range 1 and 3\n", "stmt.setFilter(Filter.range(\"bin1\", 1, 3));\n", "// expression filter to get bin1 in range 2 and 4\n", "QueryPolicy policy = new QueryPolicy(client.queryPolicyDefault);\n", "policy.filterExp = Exp.build(\n", " Exp.and(\n", " Exp.ge(Exp.intBin(\"bin1\"), Exp.val(2)),\n", " Exp.le(Exp.intBin(\"bin1\"), Exp.val(4))));\n", "RecordSet rs = client.query(policy, stmt);\n", "System.out.format(\"Query based on index predicate and expression results\\n\");\n", "while (rs.next()) {\n", " Key key = rs.getKey();\n", " Record record = rs.getRecord();\n", " System.out.format(\"key=%s bins=%s\\n\", key.userKey, record.bins);\n", "}\n", "rs.close();" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### A null set value when an index predicate is used works records belonging to no (null) set, but without an index predicate works on the entire namespace.\n", "The scope of an index is a set. An index must exist on the null set when an index predicate is used with the null set." ] }, { "cell_type": "code", "execution_count": 69, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Query with expression filter on null set results:\n", "set=sql-select-large key=id-3 bins={bin1=3, bin2=1003}\n", "set=sql-select-small key=id-3 bins={bin1=3, bin2=1003}\n", "set=null key=id-3 bins={bin1=3, bin2=1003}\n", "\n", "Scan with expression filter on null set results:\n", "set=sql-select-large key=id-3 bins={bin1=3, bin2=1003}\n", "set=sql-select-small key=id-3 bins={bin1=3, bin2=1003}\n", "set=null key=id-3 bins={bin1=3, bin2=1003}\n" ] } ], "source": [ "// query with a null set\n", "Statement stmt = new Statement();\n", "stmt.setNamespace(Namespace); \n", "stmt.setSetName(NullSet); \n", "\n", "// the filter selects records with bin1=3 in all sets\n", "QueryPolicy policy = new QueryPolicy(client.queryPolicyDefault);\n", "policy.filterExp = Exp.build(\n", " Exp.eq(Exp.intBin(\"bin1\"), Exp.val(3)));\n", "RecordSet rs = client.query(policy, stmt);\n", "System.out.format(\"Query with expression filter on null set results:\\n\");\n", "while (rs.next()) {\n", " Key key = rs.getKey();\n", " Record record = rs.getRecord();\n", " System.out.format(\"set=%s key=%s bins=%s\\n\", key.setName, key.userKey, record.bins);\n", "}\n", "rs.close();\n", "\n", "// scan with a null set\n", "public class ScanParallel implements ScanCallback {\n", "\tpublic void scanCallback(Key key, Record record) {\n", " System.out.format(\"set=%s key=%s bins=%s\\n\", key.setName, key.userKey, record.bins);\n", " }\n", "}\n", "ScanPolicy policy = new ScanPolicy();\n", "policy.filterExp = Exp.build(\n", " Exp.eq(Exp.intBin(\"bin1\"), Exp.val(3)));\n", "System.out.format(\"\\nScan with expression filter on null set results:\\n\");\n", "client.scanAll(policy, Namespace, null, new ScanParallel()); // null set" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Computed Fields with Server Function\n", "An arbitrary function registered on the server (UDF) is invoked on the specified record. In this tutorial, we deal with a single record oriented functions as opposed to \"stream oriented\" functions. The latter will be discussed in a subsequent notebook on Aggregates in this series. \n", "\n", "`SELECT func() FROM namespace.set WHERE id = key`\n", "- `Object execute(WritePolicy policy, Key key, String packageName, String functionName, Value... functionArgs)`\n", "\n", "The API returns a generic Object which can be anything like a single value or a dictionary. \n", "Note, UDFs may not be appropriate for performance sensitive applications; for record-oriented functions, simply retrieving the record and computing the function on the client site may be faster. A read-write function may be alternatively implemented atomically on the client side using the [read-modify-write pattern](../python/transactions_rmw_pattern.ipynb).\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Create User Defined Function (UDF)\n", "Examine the following Lua code that takes two bins and returns their sum and product. Create a \"udf\" directory under \"java\" and create a file \"computed_fields.lua\" with this Lua code.\n", "\n", "Add the following code to the file \"computed_fields.lua\" in the sub-directory \"udf\":\n", "\n", "
\n",
    "-- computed_fields.lua - return sum and product of specified bins\n",
    "\n",
    "function sum_and_product(rec, binName1, binName2)\n",
    "    local ret = map()           -- Initialize the return value (a map)\n",
    "    ret[binName1] = rec[binName1]\n",
    "    ret[binName2] = rec[binName2]\n",
    "    ret['sum'] = rec[binName1] + rec[binName2]\n",
    "    ret['product'] = rec[binName1] * rec[binName2]\n",
    "    return ret\n",
    "end\n",
    "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Register UDF\n", "Register the UDF with the server." ] }, { "cell_type": "code", "execution_count": 70, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Registered the UDF module computed_fields.lua." ] } ], "source": [ "import com.aerospike.client.task.RegisterTask;\n", "import com.aerospike.client.Language;\n", "\n", "String UDFDir = \"./udf\";\n", "String UDFFile = \"computed_fields.lua\";\n", "client.removeUdf(null, UDFFile);\n", "RegisterTask task = client.register(policy, UDFDir+\"/\"+UDFFile, \n", " UDFFile, Language.LUA);\n", "task.waitTillComplete();\n", "\n", "System.out.format(\"Registered the UDF module %s.\", UDFFile);;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Execute UDF \n", "Execute the UDF to retrieve the computed values." ] }, { "cell_type": "code", "execution_count": 71, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Computed fields using UDF results:\n", "key=id-1 object={product=1001, sum=1002, bin1=1, bin2=1001}\n" ] } ], "source": [ "// the UDF function returns sum and product of the specified bins\n", "Key key = new Key(Namespace, NullSet, \"id-1\");\n", "Object obj = client.execute(null, key, \"computed_fields\", \"sum_and_product\", Value.get(\"bin1\"), Value.get(\"bin2\"));\n", "System.out.format(\"Computed fields using UDF results:\\n\");\n", "System.out.format(\"key=%s object=%s\\n\", key.userKey, obj.toString());;" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Takeaways and Conclusion\n", "Many developers that are familiar with SQL would like to see how SQL operations translate to Aerospike. We looked at how to implement various SELECT statements. This should be generally useful irrespective of the reader's SQL knowledge. While the examples here use synchronous execution, many operations can also be performed asynchronously. " ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Clean up\n", "Remove tutorial data and close connection." ] }, { "cell_type": "code", "execution_count": 72, "metadata": { "ExecuteTime": { "end_time": "2020-12-29T20:49:19.972650Z", "start_time": "2020-12-29T20:49:19.967344Z" } }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Removed tutorial data and closed server connection.\n" ] } ], "source": [ "client.dropIndex(null, Namespace, SmallSet, IndexName);\n", "client.truncate(null, Namespace, null, null);\n", "client.close();\n", "System.out.println(\"Removed tutorial data and closed server connection.\");" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Further Exploration and Resources\n", "Here are some links for further exploration\n", "\n", "Resources\n", "- Related notebooks\n", " - [Queries](https://github.com/aerospike-examples/interactive-notebooks/blob/main/notebooks/python/query.ipynb)\n", " - Other notebooks in the SQL series on 1) UPDATE, CREATE, and DELETE, and 2) Aggregates.\n", "- [Aerospike Presto Connector](https://www.aerospike.com/docs/connect/access/presto/index.html)\n", "- Aerospike Developer Hub\n", " - [Java Developers Resources](https://developer.aerospike.com/java-developers)\n", "- Github repos\n", " - [Java code examples](https://github.com/aerospike/aerospike-client-java/tree/master/examples/src/com/aerospike/examples)\n", "- Documentation\n", " - [Java Client](https://www.aerospike.com/docs/client/java/index.html)\n", " - [Java API Reference](https://www.aerospike.com/apidocs/java/)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next steps\n", "\n", "Visit [Aerospike notebooks repo](https://github.com/aerospike-examples/interactive-notebooks) to run additional Aerospike notebooks. To run a different notebook, download the notebook from the repo to your local machine, and then click on File->Open, and select Upload." ] } ], "metadata": { "kernelspec": { "display_name": "Java", "language": "java", "name": "java" }, "language_info": { "codemirror_mode": "java", "file_extension": ".jshell", "mimetype": "text/x-java-source", "name": "Java", "pygments_lexer": "java", "version": "11.0.8+10-LTS" }, "toc": { "base_numbering": 1, "nav_menu": {}, "number_sections": true, "sideBar": true, "skip_h1_title": false, "title_cell": "Table of Contents", "title_sidebar": "Contents", "toc_cell": true, "toc_position": {}, "toc_section_display": true, "toc_window_display": false } }, "nbformat": 4, "nbformat_minor": 4 }