Cache service#

The Starburst Enterprise platform (SEP) cache service provides the ability to configure and automate the management of table scan redirections and materialized views in supported connectors.

The service connects to an existing SEP installation to run queries for copying data from the source catalog to the target catalog. The target catalog is regularly synchronized with the source and used as a cache.

The cache service can be run as a standalone service or within the coordinator process. You can interact with it using its REST API, or the cache service CLI.

Note

Table scan redirection and the cache service requires a valid Starburst Enterprise license.

Requirements#

The cache service has similar requirements to SEP, which are described on the Deploying page.

Linux Operating System#

64-bit required
Newer release preferred, especially when running on containers

Java Runtime Environment#

The cache service requires a 64-bit version of Java 11. Newer major versions such as Java 12 or 13 are not supported – they may work, but are not tested.

Python#

version 2.6.x, 2.7.x, or 3.x
required by the bin/launcher script only

Relational database#

The cache service requires an externally managed database for storing table scan redirections data and materialized view definitions. The following RDBMS are supported:

MySQL 8.0.12 or higher
PostgreSQL 9.6 or higher
OracleDB 12.2.0.1 or higher

Table scan redirection#

The cache service requires configured table scan redirection for each source catalog where table scans are desired.

Materialized views#

The cache service handles materialized view refreshes. To support this, each supported catalog where storage tables reside must be configured to allow the creation of these tables.

Installation#

The cache service can be deployed either as a standalone application separate from your SEP cluster or within the existing coordinator process.

Standalone deployment#

A standalone deployment ensures that the service is not affected by coordinator performance, or the deployment of a new release on the SEP cluster.

Deployment of the cache service in Kubernetes can be managed with the available Helm chart .

Manual deployment relies on using a tarball:

To download the cache service binary file, contact Starburst Support
Starburst Support provides access to a file named much like starburst-cache-service-*.tar.gz
Extract it, for example with tar xfvz starburst-cache-service-*.tar.gz

The resulting directory starburst-cache-service-nnn, with nnn replaced by the release number, is called the installation directory. It contains all necessary resources.

Move the extracted directory into the desired location, such as /opt/, and you are ready to proceed with configuring the service.

Embedded mode#

The cache service can be setup to run within the coordinator process by providing configuration for the cache service on the coordinator in etc/cache.properties. This mode of deployment does not require installation of any additional packages or running a separate service.

Keep in mind that the coordinator JVM is used, and its JVM and logging configuration is also used for the cache service.

Configuration#

Create an etc directory inside the installation directory to hold the following configuration files:

etc/jvm.config - command line configurations options for starting the Java Virtual Machine (JVM) used by the cache service
etc/config.properties - configuration for the cache service
etc/log.properties - optional logging configuration for the cache service
etc/rules.json - JSON file specifying the source tables and target connector for the cache along with the schedule for refreshing them
etc/type-mapping.json - optional JSON file to specify type mapping between source and target catalogs.

JVM configuration#

The Java Virtual Machine (JVM) config file, etc/jvm.config, contains a list of command line options used for launching the JVM running the cache service. The format of the file is a list of options, one per line. These options are not interpreted by the shell, so options containing spaces, or other special characters, should not be quoted.

The following provides a good starting point for creating etc/jvm.config:

-server
-Xmx512M
-XX:+ExitOnOutOfMemoryError
-XX:+HeapDumpOnOutOfMemoryError

An OutOfMemoryError typically leaves the JVM in an inconsistent state. The above configuration causes the JVM to write a heap dump file for debugging, and forcibly terminate the process when this occurs.

Configuration properties#

The configuration properties file, etc/config.properties, contains the configuration for the cache service, when deployed as a standalone system. Users of the embedded mode use a etc/cache.properties file on the coordinator with the same properties.

The following is a minimal configuration for the service:

service-database.user=alice
service-database.password=test123
service-database.jdbc-url=jdbc:mysql://mysql-server:3306/redirections
starburst.user=bob
starburst.jdbc-url=jdbc:trino://coordinator:8080
rules.file=etc/rules.json

The properties to configure the cache service are explained in detail in the following sections.

General cache service configuration properties#
Property name	Description
`starburst.user`	Username to connect to the SEP cluster for executing queries to refresh the cached tables. This user must also be able to read the source tables for any materialized views it refreshes.
`starburst.password`	Password to connect to the SEP cluster when password based authentication is enabled on the SEP cluster.
`starburst.jdbc-url`	JDBC URL of the SEP cluster used for executing queries to refresh the cached tables. You can use JDBC driver parameters in the connection string to configure details of the connection. For example, use `jdbc:trino://coordinator?SSL=true` for a cluster secured with TLS.
`cache-service.uri`	URI of the cache service, including the hostname and port number.
`rules.file`	Path to the JSON file containing rules for identifying source tables and target connector for caching. It also specifies a schedule for refreshing cached tables.
`rules.refresh-period`	Frequency at which cache rules are refreshed from the `rules.file`. Defaults to `1m`.
`max-table-import-threads`	Maximum number of table import jobs that can be run in parallel. Defaults to `20`.
`refresh-interval`	Frequency at which the cache service triggers refresh of cached tables and checks for a need to refresh materialized views based on their refresh interval and cron expression. Do not set this cache service interval to large values that can cause defined materialized view refreshes to be skipped. Defaults to `2m`.
`refresh-initial-delay`	Initial delay for startup of the refresh. Defaults to `0s`.
`cleanup-interval`	Frequency at which cache service triggers cleanup of expired tables in the cache. Defaults to `5m`.
`cleanup-initial-delay`	Initial delay for startup of the cleanup. Defaults to `0s`.

The following required properties allow you to configure the connectivity to the service database used for storing redirections.

Cache service database related configuration properties#
Property name	Description
`service-database.user`	Username used to connect to the database storing table redirections
`service-database.password`	Password used to connect to the database storing table redirections
`service-database.jdbc-url`	JDBC URL of the database storing table redirections, only MySQL, PostgreSQL and Oracle URLs are supported
`service-database.connection-pool.enabled`	Enables pooling for connections to the service database. Defaults to `true`.
`service-database.connection-pool.max-size`	Maximum number of connections in the pool. Defaults to `10`.
`service-database.connection-pool.idle-timeout`	Maximum time an idle connection is kept in the pool. Defaults to `10m`.

The following optional properties allow you to configure the table import configuration used when running queries on SEP to populate the cached views.

Cache service cached view import properties#
Property name	Description	Default
`unpartitioned.writer-count`	Number of writers per task when writing unpartitioned table	4
`unpartitioned.scale-writers`	Scale writers when writing unpartitioned table	false
`unpartitioned.writer-min-size`	Target minimum size of writer output when writing unpartitioned table with writers scaling	32MB
`partitioned.use-preferred-write-partitioning`	Use table partitioning to parallelize writes between worker nodes when writing to a partitioned table. This reduces import memory usage and improves cached table file sizes.	true
`partitioned.preferred-write-partitioning-min-number-of-partitions`	The minimum number of written partitions that is required to use connector `preferred write partitioning`	1
`partitioned.writer-count`	Number of writers per task when writing partitioned table	4
`partitioned.scale-writers`	Scale writers when writing partitioned table	false
`partitioned.writer-min-size`	Target minimum size of writer output when writing partitioned table with writers scaling	32MB

The following optional properties allow configuring type mapping rules:

Cache service type mapping configuration#
Property name	Description
`type-mapping`	Set the kind of type mapping to apply: `NONE` `CAST_TIMESTAMPS_TO_MILLISECONDS` `CAST_TIMESTAMPS_TO_MICROSECONDS` `CAST_TIMESTAMPS_TO_NANOSECONDS` `FILE` Defaults to `NONE`.
`type-mapping.file`	Path to the JSON file
`type-mapping.file.refresh-period`	Frequency to use to refresh the type mapping rules from the `type-mapping.file` file. Defaults to `1m`.

TLS and authentication#

File based password authentication can be configured for the cache service by adding the following properties:

http-server.https.enabled=true
http-server.authentication.type=password
http-server.https.keystore.path=etc/auth/localhost.keystore
http-server.https.keystore.key=changeit
file.password-file=etc/auth/password.db

HTTP and authentication properties#
Property name	Description	Default
`http-server.http.port`	HTTP port for the cache service	8180
`http-server.https.port`	HTTPS port of the cache service	8543
`http-server.https.enabled`	Flag to activate HTTPS/TLS	false
`http-server.authentication.type`	Authentication type used for the cache service, use `password` for password file based authentication	none
`http-server.https.keystore.path`	Path to the JKS keystore file used for TLS
`http-server.https.keystore.key`	Name of the key in the JKS keystore used for TLS
`file.password-file`	Path to the password file used with the file authentication type

Log Levels#

The optional log levels file, etc/log.properties, allows setting the minimum log level for named logger hierarchies. Every logger has a name, which is typically the fully qualified name of the class that uses the logger. Loggers have a hierarchy based on the dots in the name, like Java packages. For example, consider the following log levels file:

com.starburstdata.cache=INFO

This sets the minimum level to INFO for both com.starburstdata.cache.db and com.starburstdata.cache.rules. The default minimum level is INFO, thus the above example does not actually change anything. There are four levels: DEBUG, INFO, WARN and ERROR.

Table scan refresh and redirection rules#

Note

The information in the section applies to table scan redirections only. Materialized view management is described in our Hive connector documentation.

A JSON file, rules.json, is used to define rules to configure which tables are cached by the service, the catalog where the tables are accessed from, and the schedule for refreshing them. Cached tables can be refreshed either incrementally, or as a full refresh.

There are numerous properties available for use in creating rules discussed in this section that apply either globally, or to a specific catalog.

Catalogs can have more than one rule, but a rule can apply to only one catalog. A rule can apply to one or more schemas and one or more tables within the named catalog.

Warning

Any changes to the sources tables and their columns require updates to the caching rules that use them. This includes rules that do not specify columns, if any columns are added, removed or renamed in the source table. Such changes require that the cache is expired with the expire_redirection command from the cache service CLI to create a new redirection with the updated table definition.

Note

To avoid queries on stale data, if a cached table can not be successfully refreshed, it is not used for table scan redirection. Instead, the query executes directly against the source data.

Full refreshes#

By default, each refresh of a cached table results in all the rows from the specified columns of the source table getting loaded into the cached table. This is called a full refresh. For cached tables that are completely refreshed, the cached table has a TTL (time to live) computed as the sum of the effective maximum import duration and the duration defined by its effective grace period. This allows any running query that started just before the cached table is expired to finish gracefully. After the TTL expires, the cache service removes older cached tables once the newer tables is available.

Incremental refreshes#

When the source table is large and requires frequent refreshes, the refresh can be configured to load data from the source table incrementally. This incremental refresh approach is more performant, but not suitable for all source tables and data modifications. Specifically modifications or deletions of of existing rows are ignored.

The initial refresh for an incrementally cached table is equivalent to a full refresh. It is a bulk load of all the rows from the source table. For incrementally cached tables, the cached table is not removed until the corresponding redirection rule is removed or modified in rules.json.

Depending on the properties of the source table, incremental refresh can be set up in one of two ways:

By specifying a strictly incrementing column used to detect new rows.
By specifying a monotonically increasing column from the source table in combination with a configured predicate expression.

When using a strictly incrementing column from the source table to detect new rows, the cache service copies only the rows from source table where the value of incrementalColumn is greater than the values already present in the cached table.

The following example demonstrates this configuration:

{
  "catalogName": "mysqlevents",
  "schemaName": "web",
  "tableName": "events",
  "refreshInterval": "2m",
  "gracePeriod": "15m",
  "incrementalColumn": "event_id"
},

Fact tables often contain columns that increase monotonically. Examples are integer values for IDs, and date or datetime columns using the current time of record creation. To use such a column for incremental refreshes, you must specify a predicate expression as a filter condition. This filter is applied in addition to the filter applied on incrementalColumn by the cache service to avoid duplicating data and delay data collection until all updates to it are presumed to be complete, as shown in the following example:

{
  "catalogName": "postgresqlevents",
  "schemaName": "web",
  "tableName": "events",
  "refreshInterval": "2m",
  "gracePeriod": "15m",
  "incrementalColumn": "event_hour",
  "predicate": "event_hour < (SELECT date_add('hour', -1, latest_event_hour) FROM (SELECT max(event_hour) AS latest_event_hour FROM postgresql.foo.events))"
}

In this example, the predicate is designed to create a one hour buffer after a record is created during which the data is subject to updates. After that buffer has expired, the data is loaded and all subsequent updates are ignored.

Finally, incremental refreshes can also be assigned a predicate that removes data that is no longer needed. In the following example, events older than 31 days are removed:

{
  "catalogName": "mysqlevents",
  "schemaName": "web",
  "tableName": "events",
  "refreshInterval": "2m",
  "gracePeriod": "15m",
  "incrementalColumn": "event_id"
  "deletePredicate": "event_date < date_add('day', -31, CURRENT_DATE)"
},

Global properties#

The global properties listed in this section, when defined, are applied to all rules that do not have a specific value set. For instance, if a rule does not have a specific gracePeriod defined, the defaultGracePeriod value, if set, is used. Some properties have a default value that is used when the property does not exist in the rules.json file.

Global rules properties#
Property name	Description
defaultGracePeriod	The duration which a previous version of a cached table is retained so that any active queries using it may complete, once the latest import is available. Can be overridden within a specific rule using `gracePeriod`. Defaults to `10m`.
defaultMaxImportDuration	The maximum allowed execution time for a query used to populate a cached table. Can be overridden within a specific rule using `maxImportDuration`. Must be smaller than the refresh interval, and greater than or equal to 1m. Note that this field is required when using `cronExpression` to define a refresh schedule.
defaultCacheCatalog	The default catalog that configures the data source where the cached tables are stored. Can be overridden within a specific rule using `cacheCatalog`.
defaultCacheSchema	The default schema where cached tables are stored. You must ensure that the schema exists and that the cache service user has permissions to read and write to it. Can be overridden within a specific rule using `cacheSchema`.
defaultUnpartitionedImportConfig	A collection of import properties applied by default when writing unpartitioned cached tables named in the rule. Uses nested importConfig settings properties.
defaultPartitionedImportConfig	A collection of import properties applied by default when writing partitioned cached tables named in the rule. Uses nested importConfig settings properties.
cleanup-interval	Interval at which the cache service removes stale cached tables and runs cleanup queries for rules where a `deletePredicate` is defined. Defaults to `5m`.

The following example shows these properties used in a rules.json file:

{
  "defaultGracePeriod": "42m",
  "defaultMaxImportDuration": "1m",
  "defaultCacheCatalog": "default_cache_catalog",
  "defaultCacheSchema": "default_cache_schema",
  "defaultUnpartitionedImportConfig": {
    "usePreferredWritePartitioning": false,
    "preferredWritePartitioningMinNumberOfPartitions": 1,
    "writerCount": 128,
    "scaleWriters": false,
    "writerMinSize": "110MB"
  },
  "defaultPartitionedImportConfig": {
    "usePreferredWritePartitioning": true,
    "preferredWritePartitioningMinNumberOfPartitions": 40,
    "writerCount": 256,
    "scaleWriters": false,
    "writerMinSize": "52MB"
  }
}

Rule-specific properties#

The rule-specific properties listed in this section override a similar available global property. For instance, if a rule sets a specific maxImportDuration, the global defaultMaxImportDuration value is ignored.

Rule-specific properties#
Property name	Description
catalogName	The name of the source catalog. Required.
cacheCatalog	The catalog that defines the data source where the cached tables are stored.
cacheSchema	The schema where the cached tables are stored. You must ensure that the schema exists and that the cache service user has permissions to read and write to it.
schemaName	The specific schema to use within a rule. You must specify this or `schemaNameLike`, but not both.
schemaNameLike	SQL `LIKE` expression to match multiple schema names. You must specify this or `schemaName`, but not both.
tableName	The specific table to use within a rule. You must specify this or `tableNameLike`, but not both.
tableNameLike	SQL `LIKE` expression to match multiple table names. You must specify this or `tableName`, but not both.
refreshInterval	The time duration between refreshes. Cannot be used with `cronExpression`.
cronExpression	A cron expression defining the refresh schedule. Cannot be used with `refreshInterval`.
gracePeriod	The duration which a previous version of a cached table is retained so that any active queries using it may complete. Overrides the global `defaultGracePeriod` if set.
maxImportDuration	Overrides the global `defaultMaxImportDuration` if set. Must be smaller than the refresh interval, and greater than or equal to 1m. Note that this field is required when using `cronExpression` to define a refresh schedule.
columns	Comma-separated list of columns to be imported. If not specified, all available columns are imported. Note that redirections can only be used when all columns required by a given table scan are present.
partitionColumns	Comma-separated list of columns to use for partitioning.
bucketColumns	Comma-separated list specifying the columns to be bucketed. Requires the use of `bucketCount`.
bucketCount	Defines the number of buckets for data in the specified columns.
sortColumns	Comma separated list specifying the columns to sort on. Requires the use of `bucketColumns`.
incrementalColumn	Required for incremental refresh; if not specified, a full refresh is performed. This column is used by the service to apply a `incrementalColumn > (SELECT max(incrementalColumn) FROM cachedTable)` filter when loading data incrementally from the source table. This facilitates loading only newer data from the source table instead of the entire table in each refresh iteration.
predicate	SQL expression used in conjunction with `incrementalColumn` to delay loading data until all presumed updates are complete.
deletePredicate	Optionally specifies a predicate clause to clean up old data from the cached table. It is used by the cache service to run a `DELETE FROM cachedTable WHERE <deletePredicate>` query at every `cleanup-interval`. Available where connector used to store the cached table supports `DELETE FROM` statements.
importConfig	Overrides the default import config (partitioned or unpartitioned, depending on the use of `partitionColumns`) when loading the cached tables. If there is no `incrementalColumn` specified, only importConfig applies. If `incrementalColumn` is specified, then importConfig applies for the first bulk import, then subsequent incremental imports use `incrementalImportConfig`. Uses the nested fields described in importConfig settings.
incrementalImportConfig	Overrides the default import config (partitioned or unpartitioned, depending on the use of `partitionColumns`) when loading the cached tables during incremental imports. Uses the nested fields described in importConfig settings.

The following example shows rules defined for several catalogs, using a variety of options:

{
  "catalogName": "maketing_campaigns",
  "schemaNameLike": "apac",
  "tableNameLike": "organic",
  "refreshInterval": "60m"
  "gracePeriod": "15m",
  "incrementalColumn": "event_id",
  "deletePredicate": "event_date < date_add('day', -31, CURRENT_DATE)"
},
{
  "catalogName": "sales",
  "schemaNameLike": "contacts",
  "tableNameLike": "leads",
  "cronExpression": "* * * 2 *"
  "gracePeriod": "15m",
  "incrementalColumn": "event_hour",
  "predicate": "event_hour < (SELECT date_add('hour', -1, latest_event_hour) FROM (SELECT max(event_hour) AS latest_event_hour FROM postgresql.foo.events))"
},
{
  "catalogName": "telemetry",
  "schemaName": "allproducts",
  "tableName":  "events",
  "columns": [
    "pingtime",
    "avgMem",
    "apiCalls"
  ],
  "partitionColumns": [
    "pingtime"
  ],
  "bucketColumns": [
    "avgMem"
  ],
  "bucketCount": 5,
  "sortColumns": [
    "apiCalls"
  ],
  "refreshInterval": "123h",
  "gracePeriod": "80m",
  "maxImportDuration": "67h",
  "cacheCatalog": "ctp_table_catalog",
  "cacheSchema": "ctp_table_schema",
  "importConfig": {
    "usePreferredWritePartitioning": true,
    "preferredWritePartitioningMinNumberOfPartitions": 4,
    "writerCount": 32,
    "scaleWriters": false,
    "writerMinSize": "100MB"
  },
  "incrementalImportConfig": {
    "usePreferredWritePartitioning": false,
    "preferredWritePartitioningMinNumberOfPartitions": 1,
    "writerCount": 4,
    "scaleWriters": false,
    "writerMinSize": "100MB"
  }
}

`importConfig` properties#

The global properties for defaultPartitionedImportConfig and defaultUnpartitionedImportConfig, as well as the rule-specific importConfig and incrementalImportConfig properties use the following sub-properties:

`importConfig` properties#
Property name	Description
usePreferredWritePartitioning	Implements use-preferred-write-partitioning.
preferredWritePartitioningMinNumberOfPartitions	Implements preferred-write-partitioning-min-number-of-partitions.
writerCount	Implements task.writer-count.
scaleWriters	Implements scale-writers.
writerMinSize	Implements writer-min-size.

Type mapping rules#

Type mapping overcomes missing type support in target storage catalogs for cache service-manged table scan redirections and materialized views with Hive as the target catalog. It allows your users to create materialized views without the need to perform data type casting, and allows data engineers to create cached table projections in target catalogs where there is not a one-to-one type mapping. Type mapping uses definitions in a JSON file, type-mapping.json, to define type casting rules used when the target catalog is created or updated.

Mappings are key-value pairs as source: target. The following example shows how to construct type mapping rules for the target catalog, myhivesalesdata:

{
  "rules": {
    "myhivesalesdata": {
      "timestamp(0)": "timestamp(3)",
      "timestamp(1)": "timestamp(3)",
      "timestamp(2)": "timestamp(3)"
    }
  }
}

Each target catalog has a separate entry in the JSON file. The Trino name of any type that is supported by the source catalog can be used as a key in the type map. Values must be Trino types supported by the target catalog.

Type casting is applied to regular columns, partition columns, and the column used for incremental update.

Type mapping behavior#

Columns can only be cast based on their type. It is impossible to cast one column of a given type without casting all columns of that same type.

There are pre-configured type mappings CAST_TIMESTAMPS_TO_MILLISECONDS, CAST_TIMESTAMPS_TO_MICROSECONDS and CAST_TIMESTAMPS_TO_NANOSECONDS that can only be set using cache properties, not via the JSON type mapping. These mappings extend the precision of all timestamp-related types, they never truncate data, and ignore the target catalog name.

Some mappings, like integer -> varchar, can change the semantics of the max() function used for calculating the contents of an incremental update. For example "-1" > "7" because string lengths are compared first, so in incremental update the row with "-1" is added even if the table already contains data up to "7".

Using timestamp columns for partitioning is strongly discouraged. Timestamp partitions are silently changed to timestamp(3) while being written. This behavior can result in loss of precision for some columns. Additionally, it can lead to a huge amount of partitions, which negatively impacts performance.

Running the cache service#

The installation directory contains the launcher script in bin/launcher. It service can be started as a daemon by running the following:

bin/launcher start

Alternatively, it can be run in the foreground, with the logs and other output written to stdout/stderr. Both streams should be captured if using a supervision system like daemontools:

bin/launcher run

Run the launcher with --help to see the supported commands and command line options. In particular, the --verbose option is very useful for debugging the installation.

The launcher configures default values for the configuration directory etc, configuration files, the data directory var, and log files in the data directory. You can change these values to adjust your usage to any requirements, such as using a directory outside the installation directory, specific mount points or locations, and even using other file names.

After starting the cache service, you can find log files in the log directory inside the data directory var:

launcher.log: This log is created by the launcher and is connected to the stdout and stderr streams of the server. It contains a few log messages that occur while the server logging is being initialized, and any errors or diagnostics produced by the JVM.
server.log: This is the main log file used by the service. It typically contains the relevant information if the server fails during initialization. It is automatically rotated and compressed.
http-request.log: This is the HTTP request log which contains every HTTP request received by the server. It is automatically rotated and compressed.

JMX metrics#

Metrics about table import are reported in the JMX table jmx.current."com.starburstdata.cache:name=TableImportService".

Metrics about cached table cleanup are reported in the JMX table jmx.current."com.starburstdata.cache:name=CleanupService".

Metrics about redirections requests on the web service resources are reported in the JMX table jmx.current."com.starburstdata.cache.resource:name=RedirectionsResource".

Metrics about table import and expiration requests on the web service resource are reported in the JMX table jmx.current."com.starburstdata.cache.resource:name=CacheResource".

Cache service#

Requirements#

Linux Operating System#

Java Runtime Environment#

Python#

Relational database#

Table scan redirection#

Materialized views#

Installation#

Standalone deployment#

Embedded mode#

Configuration#

JVM configuration#

Configuration properties#

TLS and authentication#

Log Levels#

Table scan refresh and redirection rules#

Full refreshes#

Incremental refreshes#

Global properties#

Rule-specific properties#

importConfig properties#

Type mapping rules#

Type mapping behavior#

Running the cache service#

JMX metrics#

`importConfig` properties#