Built-in Connectors

This document introduces the built-in data connectors in SQLRec and their usage.

Overview

SQLRec provides multiple built-in connectors for connecting to different data storage systems. Connectors are implemented based on Calcite table abstractions, supporting SQL queries and data write operations.

If you are new to creating and using data-source tables, start with Connecting Data Sources.

Row Cache Scope

For Redis, JDBC, and MongoDB, cache-ttl / max-cache-size control a primary-key row cache on each Calcite table object, rather than a cache of complete SQL results; setting either to 0 disables it. Redis primary-key equality (including a key predicate inside AND), standalone JDBC/MongoDB primary-key equality, and KV Joins on the right table's primary key use this cache. JDBC/MongoDB non-key, range, compound, and full-scan queries do not. Neither do Milvus, filesystem, Kafka, or Flink Redis Lookup. SQL writes invalidate relevant keys on the current table object; external and Flink writes do not proactively invalidate them, so cached values can remain until TTL expiry. See Architecture: Connector Row Cache and Query Paths for the full query matrix and FLUSH behavior.

Built-in Connectors

1. Redis Connector

The Redis connector is used to connect to Redis databases, supporting key-value storage and queries.

Connector Identifier: redis

Inheritance Type: SqlRecKvTable

Features:

• Supports standalone and cluster modes
• Supports json, list, and string data structures
• Supports JSON and Protobuf value formats
• Supports local cache for query acceleration
• Supports primary key filter queries
• Supports INSERT, UPDATE, and DELETE

Configuration Parameters:

Parameter	Type	Default	Description
url	String	-	Redis connection URL, format: redis://password@host:port/db
redis-mode	String	single	Redis mode, options: single (standalone), cluster
data-structure	String	json	Data structure, options: json, list, string
format	String	json	Value format: json or protobuf
protobuf.message-class-name	String	-	Fully qualified generated Protobuf Message class; required for format = protobuf
max-list-size	Integer	0	Maximum list length, 0 means unlimited
ttl	Integer	2592000	Key expiration time (seconds), default 30 days
cache-ttl	Integer	30	Local cache expiration time (seconds), 0 means no cache
max-cache-size	Integer	100000	Maximum cached primary-key entries; 0 disables caching

Usage Example:

CREATE TABLE user_table (
  id BIGINT,
  name STRING,
  country STRING,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'url' = 'redis://localhost:6379/0'
);

CREATE TABLE user_interest_category1 (
  user_id BIGINT,
  category1 STRING,
  score FLOAT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'data-structure' = 'list',
  'url' = 'redis://localhost:6379/0'
);

CREATE TABLE user_proto (
  value STRING,
  PRIMARY KEY (value) NOT ENFORCED
) WITH (
  'connector' = 'redis',
  'url' = 'redis://localhost:6379/0',
  'format' = 'protobuf',
  'protobuf.message-class-name' = 'com.google.protobuf.StringValue'
);

Notes:

• Queries require a primary-key equality condition (for example, WHERE user_id = 42); other columns can be filtered after the key lookup, but full-table scans are unsupported
• Using local cache can significantly improve query performance
• In list mode, the primary key is a Redis list lookup key, not a unique row identifier. Multiple rows can share one key. INSERT prepends rows, and max-list-size limits the list length after writes
• In list mode, UPDATE selects old rows using the SQL predicate, removes one exact list entry per selected row, then prepends each new row. Changing the primary key moves the row to the new key. A missing old row is not inserted
• In list mode, DELETE removes all list entries identical to a selected row, while UPDATE removes one matching entry for each selected old row. The remove and insert commands in UPDATE are not atomic
• Protobuf format requires the generated Message class on the SQLRec and Flink runtime classpaths
• format = protobuf cannot be combined with data-structure = string

2. Milvus Connector

The Milvus connector is used to connect to Milvus vector databases, supporting vector similarity retrieval.

Connector Identifier: milvus

Inheritance Type: SqlRecKvTable (implements VectorSearchable)

Features:

• Supports vector similarity search (ANN)
• Supports primary key queries
• Supports filter conditions
• Supports data insertion and deletion
• Supports projection column optimization

Configuration Parameters:

Parameter	Type	Default	Description
url	String	-	Milvus server address
token	String	-	Milvus authentication token
database	String	default	Database name
collection	String	-	Collection name
batch-size	Integer	4096	Batch size for bulk insert
pool.max-idle-per-key	Integer	10	Maximum idle connections per key in the connection pool
pool.max-total-per-key	Integer	100	Maximum total connections per key in the connection pool
pool.max-total	Integer	100	Maximum total connections in the connection pool
pool.max-block-wait-duration	Long	5	Maximum blocking wait time (seconds) when acquiring a connection from the pool
pool.min-evictable-idle-duration	Long	10	Minimum evictable idle duration (seconds) for connections in the pool
flush-interval	Long	1	Flush interval (seconds) for bulk writes, flushed when the buffer is full or the interval is reached
rpc-deadline-ms	Long	30000	Timeout (milliseconds) for Milvus gRPC calls, 0 means no limit

Usage Example:

CREATE TABLE item_embedding (
  id BIGINT,
  embedding ARRAY<FLOAT>,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'milvus',
  'url' = 'http://localhost:19530',
  'token' = 'root:Milvus',
  'database' = 'default',
  'collection' = 'item_embedding'
);

Notes:

• Milvus connector supports complex filter conditions
• Vector search requires specifying the vector field and query vector
• Supports projection column optimization, returning only needed columns

3. Kafka Connector

The Kafka connector is used to connect to Apache Kafka message queues, supporting message writing.

Connector Identifier: kafka

Inheritance Type: SqlRecTable

Features:

• Supports message writing to Kafka Topic
• Supports JSON and Protobuf messages
• Supports batch sending optimization

Configuration Parameters:

Parameter	Type	Default	Description
properties.bootstrap.servers	String	-	Kafka Broker address
topic	String	-	Kafka Topic name
format	String	json	Message format: json or protobuf
protobuf.message-class-name	String	-	Fully qualified generated Protobuf Message class; required for format = protobuf
properties.producer.linger.ms	Integer	5000	Batch sending wait time (milliseconds)

Usage Example:

CREATE TABLE rec_log_kafka (
  user_id BIGINT,
  item_id BIGINT,
  item_name STRING,
  rec_reason STRING,
  req_time BIGINT,
  req_id STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'rec_log',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

Notes:

• Kafka connector is mainly used for message writing, does not support query operations
• linger.ms parameter controls batch sending, larger values can improve throughput but increase latency
• Protobuf format requires the generated Message class on SQLRec's runtime classpath

4. JDBC Connector

The JDBC connector is used to connect to relational databases (e.g., PostgreSQL, MySQL), supporting SQL queries and data writes.

Connector Identifier: jdbc

Inheritance Type: SqlRecKvTable

Features:

• Supports various JDBC databases (PostgreSQL, MySQL, etc.)
• Supports primary key queries and local cache acceleration
• Supports complex filter condition queries (not limited to primary key filtering)
• Supports data upsert and deletion
• Uses HikariCP connection pool for database connection management
• Supports custom JDBC properties

Configuration Parameters:

Parameter	Type	Default	Description
url	String	-	JDBC connection URL, e.g. jdbc:postgresql://host:port/db
table-name	String	-	JDBC table name
username	String	""	Database username
password	String	""	Database password
driver	String	""	JDBC driver class name, e.g. org.postgresql.Driver
schema	String	""	Database schema name (e.g. PostgreSQL schema)
max-cache-size	Integer	100000	Maximum cached primary-key entries; 0 disables caching
cache-ttl	Integer	30	Local cache expiration time (seconds), 0 means no cache
connection.pool.size	Integer	0	Connection pool max size (HikariCP maximumPoolSize), 0 means use default
connection.pool.min-idle	Integer	0	Connection pool min idle connections, 0 means use default
connection.pool.idle-timeout	Long	0	Connection pool idle timeout in seconds, 0 means use default
connection.pool.max-lifetime	Long	0	Connection pool max lifetime in seconds, 0 means use default
connection.pool.connection-timeout	Long	0	Connection pool connection timeout in seconds, 0 means use default
connection.pool.validation-timeout	Long	0	Connection pool validation timeout in seconds, 0 means use default
connection.pool.keepalive-time	Long	0	Connection pool keepalive time in seconds, 0 means use default
connection.pool.pool-name	String	""	Connection pool name
jdbc.properties.*	String	-	Custom JDBC properties, the part after jdbc.properties. prefix is used as the property name

Usage Example:

CREATE TABLE user_profile (
  id BIGINT,
  name STRING,
  age INT,
  country STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:postgresql://localhost:5432/mydb',
  'table-name' = 'user_profile',
  'username' = 'postgres',
  'password' = 'postgres',
  'driver' = 'org.postgresql.Driver'
);

Notes:

• JDBC connector supports complex filter conditions, not limited to primary key equality filtering
• Uses HikariCP connection pool for database connection management, sharing the pool for the same URL and username
• Supports upsert operations, automatically determining insert or update based on primary key
• Custom JDBC properties can be passed via the jdbc.properties.* prefix

5. MongoDB Connector

The MongoDB connector is used to connect to MongoDB document databases, supporting document queries and data writes.

Connector Identifier: mongodb

Inheritance Type: SqlRecKvTable

Features:

• Supports MongoDB connection URI
• Supports primary key queries and local cache acceleration
• Supports complex filter conditions (AND, OR, comparison operators, IS NULL, etc.)
• Supports data upsert and deletion
• Automatically pushes down Calcite filter conditions to MongoDB queries

Configuration Parameters:

Parameter	Type	Default	Description
uri	String	-	MongoDB connection URI, e.g. mongodb://host:port
database	String	-	Database name
collection	String	-	Collection name
max-cache-size	Integer	100000	Maximum cached primary-key entries; 0 disables caching
cache-ttl	Integer	30	Local cache expiration time (seconds), 0 means no cache

Usage Example:

CREATE TABLE user_behavior (
  event_id BIGINT,
  user_id BIGINT,
  item_id BIGINT,
  action STRING,
  timestamp BIGINT,
  PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'connector' = 'mongodb',
  'uri' = 'mongodb://localhost:27017',
  'database' = 'recommendation',
  'collection' = 'user_behavior'
);

Notes:

• MongoDB connector supports complex filter conditions including AND, OR, equals, not equals, greater than, less than, etc.
• Filter conditions that cannot be pushed down will be handled by Calcite in memory
• MongoClient instances are shared for the same URI
• Upsert operations automatically determine insert or update based on primary key
• Behavior records use event_id as the primary key; using user_id would make later writes for the same user replace an existing record

6. Filesystem Connector

The Filesystem connector is used to read data files from the local file system, supporting CSV and JSON formats.

Connector Identifier: filesystem

Inheritance Type: SqlRecKvTable

Features:

• Supports CSV and JSON file formats
• Data is loaded only once on first access; subsequent accesses use in-memory data
• Supports primary key queries and filter queries
• Uses the primary key as a lookup key and retains multiple rows per key
• Supports in-memory inserts, updates, and deletes (does not write back to the file system)
• Automatically initializes as an empty table if no path is configured or the path does not exist

Configuration Parameters:

Parameter	Type	Default	Description
path	String	-	File path, supports file:/// prefix, e.g. file:///path/to/data.csv
format	String	csv	File format, options: csv, json

Usage Example:

-- CSV format
CREATE TABLE user_profile (
  id INT,
  name STRING,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'filesystem',
  'path' = '/data/users.csv',
  'format' = 'csv'
);

-- JSON format
CREATE TABLE product_info (
  id INT,
  name STRING,
  price INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'filesystem',
  'path' = '/data/products.json',
  'format' = 'json'
);

-- No path specified, initializes as empty table
CREATE TABLE temp_table (
  id INT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'filesystem'
);

-- Multiple interests can share one user ID
CREATE TABLE user_interest (
  user_id BIGINT,
  category STRING,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'filesystem',
  'path' = '/data/user_interest.csv'
);

Notes:

• The first line of a CSV file is treated as a header and is automatically skipped
• CSV files support quoted fields, commas inside fields, and escaped double quotes; newlines inside fields are not supported
• JSON files support array format [{...}, {...}] and single object format {...}
• Write operations only modify in-memory data and do not write back to the file system
• If the path does not exist or the format is invalid, the table is initialized as an empty table without throwing an exception
• All CSV/JSON rows with the same key are retained; writes prepend a row and deletes remove every matching full row
• UPDATE replaces each selected old row; changing its lookup key preserves other rows under the old key
• A single-column primary key is still required, but it is only a lookup key and need not be unique per row