# Hazelcast as a data source

> How to read and write data to Hazelcast as a data source

Source: https://orbitalhq.com/docs/describing-data-sources/hazelcast

In addition to [using Hazelcast as a cache](/docs/querying/caching#using-an-external-cache), Orbital
can use Hazelcast as a source for reading and writing data, as well as a streaming data source.

## Adding a Hazelcast data source
To enable Hazelcast as a data source, first [define a connection](/docs/describing-data-sources/configuring-connections) in your `connections.conf` file.

The simplest configuration simply needs a list of addresses to connect to:

```hocon connections.conf
hazelcast {
   myHazelcast {
      connectionName = myHazelcast
      addresses = ["localhost:5701"]
   }
}
```

### Advanced configuration
Hazelcast has a number of connection parameters, which you may wish to tune. You can specify a full Hazelcast client configuration file, and provide the path.

For [XML](https://github.com/hazelcast/hazelcast/blob/master/hazelcast/src/main/resources/hazelcast-client-default.xml) based configuration:

```hocon
hazelcast {
   prodHazelcastCluster {
      connectionName = myHazelcast
      xmlConfig="hazelcast-config.xml"
   }
}
```

If your Hazelcast client config is in [yaml](https://github.com/hazelcast/hazelcast/blob/master/hazelcast/src/main/resources/hazelcast-client-default.yaml) format, you can define it your connection as:

```hocon
hazelcast {
   prodHazelcastCluster {
      connectionName = myHazelcast
      yamlConfig="hazelcast-config.yaml"

   }
}
```

Paths can be either absolute (located on the server where Orbital is deployed), or relative (resolved relative to the `conncections.conf` file declaring the Hazelcast connection).

The Hazelcast docs provide more information on [declarative configuration settings](https://docs.hazelcast.com/hazelcast/5.4/configuration/configuring-declaratively)
and provide [full examples](https://github.com/hazelcast/hazelcast/blob/master/hazelcast/src/main/resources/hazelcast-client-full-example.xml).

## Supported formats
When values are written to Hazelcast, they must be serialized.

The following formats are supported. Note that because Orbital does not
have access to custom Java classes, serialization formats that require classloading
(such as `Serializable`, `Externalizable` and Custom Serialization) are not supported

### Compact (preferred)
[Compact Serialization](https://docs.hazelcast.com/hazelcast/5.3/serialization/compact-serialization) is the preferred
format when using Orbital to read/write to Hazelcast.

To enable Compact serialization, add a `@CompactObject` annotation to your Taxi definition:

```taxi
>  import com.orbitalhq.hazelcast.CompactObject
   import com.orbitalhq.hazelcast.HazelcastMap

   @HazelcastMap(name = "films")
>  @CompactObject
   closed model Film {
      @Id
      filmId : FilmId inherits Int
      // ...omitted...
   }
```

### JsonValue 
Hazelcast's [JsonValue](https://docs.hazelcast.com/hazelcast/5.3/serialization/serializing-json) is supported
for working with existing caches that have stored data using JsonValue.

Note - if you are both writing and reading the data using Orbital, it is recommended to use [Compact](#compact-preferred), 
which has better performance.

## Writing data to Hazelcast

To write data into a Hazelcast Map, define a Hazelcast Service in your taxi project,
with a write operation:

```taxi
import com.orbitalhq.hazelcast.HazelcastService
import com.orbitalhq.hazelcast.HazelcastMap
import com.orbitalhq.hazelcast.UpsertOperation
import com.orbitalhq.hazelcast.CompactObject

// The HazelcastService annotation should specify the 
// name of the connection defined in your connections.conf
@HazelcastService(connectionName = "myHazelcastConnection")
service HazelcastService {

   // Expose a write operation, and annotate it 
   // with UpsertOperation, to define the writing behaviour
   @UpsertOperation
   write operation upsert(Film):Film
}

type Language inherits String

// Define the name of the map to use. If not present in Hazelcast,
// the map wil be created
@HazelcastMap(name = "films")
// specify the serialization format 
@CompactObject
closed model Film {
   // A field annotated with @Id becomes the key.
   // Composite keys are not supported
   @Id
   filmId : FilmId inherits Int
   title : Title inherits String
   languages: Language[]
}
```

With your map declared, use a mutation query to insert data:

### Examples

#### Inserting a static value into Hazelcast
```taxi
// inserting a static value into Hazelcast
given { film : Film = 
  {
    filmId : 123,
    title : "Star Wars",  
    languages : ["English" , "American" ]
  } 
}
call HazelcastService::upsert
```

#### Storing a Query Result in Hazelcast
To write the outcome of a query to Hazelcast, it's not 
necessary for the query's result format to align with the format of the 
persisted value. 

Orbital will automatically adapt the query result to the 
required persistence format, which may involve projections and even 
calling additional services if needed.

```taxi
find { NetflixFilms[] } as {
  // projection not shown
}
call HazelcastService::upsert
```

#### Streaming data into Hazelcast
```taxi
stream { FilmUploadedEvent }
// Each FilmUploadedEvent is projected into a Film
// which is the input parameter to `upsert`
call HazelcastService::upsert
```

## Querying data from Hazelcast
Orbital supports querying from Hazelcast using both direct key
lookups, and rich query criteria.

### Defining a map to query
To query a map, you first define a `service` that exposes your
Hazelcast Map.

Maps are exposed using taxi's `table` declaration, as this indicates
a data source that supports rich querying.

Here's a complete example:

```taxi
import com.orbitalhq.hazelcast.HazelcastService
import com.orbitalhq.hazelcast.HazelcastMap
import com.orbitalhq.hazelcast.CompactObject

// The HazelcastService annotation should specify the 
// name of the connection defined in your connections.conf
@HazelcastService(connectionName = "myHazelcastConnection")
service HazelcastService {

    // Table is a shorthand to declare
    // a data source that supports rich querying.
    table films : Film[]
}

// Define the name of the map to query.
@HazelcastMap(name = "films")
// specify the serialization format
@CompactObject
closed model Film {
   // A field annotated with @Id becomes the key, which 
   // is used when performing key lookups
   @Id
   filmId : FilmId inherits Int
   title : Title inherits String
   languages: Language[]
}
```

### Writing queries
Once a Hazelcast map is exposed, it can be queried as a standard
data source, including used as a data source when projecting and
joining data from other sources (such as APIs, Kafka topics, or Databases).

Here are some sample queries:

#### Fetching everything from a map
```taxi
find { Film[] }
```

#### Fetching a value with a specific key
If criteria is defined against the key (as defined using an `@Id` annotation),
then a key lookup is performed:

```taxi
// Assuming FilmId is annotated as @Id in the Film model
// as shown...
model Film {
   @Id
   filmId : FilmId inherits Int
   // ..snip..
}

// Elsewhere, writing a query...
find { Film( FilmId === 123 ) }
```

#### Fetching values using criteria 

```taxi
// find all films with a FilmId < 105
find { Film[]( FilmId < 105 ) }

// find all films released after 2019 with the title Star Wars
find { Film[]( ReleaseYear < 2019 && Title == "Star Wars" ) }
```

### Collection options

_Available since 0.38.0_

You can limit, paginate and sort results using [collection options](/docs/querying/collection-options), written alongside
your filter criteria:

```taxi
find { Film[]( orderBy: Title asc, limit: 10 ) }
```

For Hazelcast, only `limit` is pushed down - applied as a paging predicate over the map, which returns entries in **key
order**. `offset`, `orderBy` and `uniqueBy` are applied by Orbital after fetching. Because the pushed-down `limit` is
key-ordered, an `orderBy` is always honoured engine-side rather than being pushed to Hazelcast.

| Option     | Behaviour against Hazelcast                   |
|------------|-----------------------------------------------|
| `limit`    | Pushed down as a key-ordered paging predicate.|
| `offset`, `orderBy`, `uniqueBy` | Applied by Orbital after fetching.       |

When a query carries an `orderBy` and Orbital is running with distributed (Hazelcast) projection, the ordered query's
projection runs locally, to preserve the ordering.

On [streaming Hazelcast queries](#streaming-data-from-hazelcast), only `limit` is supported - `orderBy`, `offset` and
`uniqueBy` are rejected, as they have no meaning over an unbounded stream.

See [collection options](/docs/querying/collection-options) for the execution model, and the
[Taxi language reference](https://taxilang.org/docs/taxiql/querying#collection-options) for the full syntax.

## Streaming data from Hazelcast
Hazelcast Map's can be treated as data streams, where inserts or updates
are created as streams of events which can be queried using Orbital.

The following events trigger the current state of the record
to be written to the event stream:

 * Entry Added
 * Entry Updated

### Declaring a Map as a Stream
To stream updates from a map, you first define a `service` that exposes your
Hazelcast Map as stream

Here's a complete example:

```taxi
import com.orbitalhq.hazelcast.HazelcastService
import com.orbitalhq.hazelcast.HazelcastMap
import com.orbitalhq.hazelcast.CompactObject

// The HazelcastService annotation should specify the 
// name of the connection defined in your connections.conf
@HazelcastService(connectionName = "myHazelcastConnection")
service HazelcastService {
   stream films : Stream<Film>
}

// Define the name of the map to query.
@HazelcastMap(name = "films")
// specify the serialization format
@CompactObject
closed model Film {
   // A field annotated with @Id becomes the key, which 
   // is used when performing key lookups
   @Id
   filmId : FilmId inherits Int
   title : Title inherits String
   languages: Language[]
}
```

### Writing streaming queries
Once a Hazelcast map is exposed as a stream, it can be queried as a standard
data source, including used as a data source when projecting and
joining data from other sources (such as APIs, Kafka topics, or Databases).

Below are some sample queries.
It's useful when testing to combine this with [writing data to Hazelcast](#writing-data-to-hazelcast) 
to trigger change events which produce values on the data stream. 

Here are some sample queries:

#### Stream all updates from a map

```taxi
stream { Film }
```

#### Stream all updates from a map, and enrich with data from other sources

```taxi
stream { Film } as {
  id : FilmId
  reviewScore : FilmReviewScore // not present in the map, will be looked up from another data source
}[]
```

#### Stream only specific events from a map
```taxi
// Only provide updates on Films whose FilmId is less than 300
stream { Film.filterEach( (FilmId) -> FilmId < 300  ) }
```