diff --git a/build.gradle.kts b/build.gradle.kts index 33ff8905c80..ccb41ba9056 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -208,7 +208,7 @@ val javadocAggregateIncludingTests by tasks.registering(Javadoc::class) { val adaptersForSqlline = listOf( ":arrow", ":babel", ":cassandra", ":druid", ":elasticsearch", - ":file", ":geode", ":innodb", ":kafka", ":mongodb", + ":file", ":geode", ":innodb", ":kafka", ":kvrocks", ":mongodb", ":pig", ":piglet", ":plus", ":redis", ":server", ":spark", ":splunk") val dataSetsForSqlline = listOf( diff --git a/kvrocks/build.gradle.kts b/kvrocks/build.gradle.kts new file mode 100644 index 00000000000..b5ddb2a7495 --- /dev/null +++ b/kvrocks/build.gradle.kts @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +dependencies { + api(project(":core")) + api(project(":linq4j")) + api("redis.clients:jedis") + + implementation("com.fasterxml.jackson.core:jackson-core") + implementation("com.fasterxml.jackson.core:jackson-databind") + implementation("com.google.guava:guava") + implementation("org.apache.calcite.avatica:avatica-core") + implementation("org.apache.commons:commons-pool2") + implementation("org.slf4j:slf4j-api") + + testImplementation(project(":testkit")) + testImplementation("org.mockito:mockito-core") + testRuntimeOnly("org.apache.logging.log4j:log4j-slf4j-impl") + testImplementation("org.testcontainers:testcontainers") +} diff --git a/kvrocks/gradle.properties b/kvrocks/gradle.properties new file mode 100644 index 00000000000..5c5c507c442 --- /dev/null +++ b/kvrocks/gradle.properties @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to you under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +description=Kvrocks adapter for Calcite +artifact.name=Calcite Kvrocks diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksConfig.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksConfig.java new file mode 100644 index 00000000000..90feccaa747 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksConfig.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Connection configuration for the Kvrocks adapter. + * + *

Holds all parameters needed to establish a connection to a Kvrocks + * instance: host, port, database index, and optional password. + */ +public class KvrocksConfig { + private final String host; + private final int port; + private final int database; + private final @Nullable String password; + + KvrocksConfig(String host, int port, int database, + @Nullable String password) { + this.host = host; + this.port = port; + this.database = database; + this.password = password; + } + + public String getHost() { + return host; + } + + public int getPort() { + return port; + } + + public int getDatabase() { + return database; + } + + public @Nullable String getPassword() { + return password; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataFormat.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataFormat.java new file mode 100644 index 00000000000..3b86b6c83f4 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataFormat.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * How values stored in Kvrocks should be interpreted as relational rows. + */ +public enum KvrocksDataFormat { + + /** Treat each value as a single raw string column. */ + RAW("raw"), + + /** Split each value by a delimiter to extract columns. */ + CSV("csv"), + + /** Parse each value as a JSON object and extract fields by mapping. */ + JSON("json"); + + private final String typeName; + + KvrocksDataFormat(String typeName) { + this.typeName = typeName; + } + + public static @Nullable KvrocksDataFormat fromTypeName(String typeName) { + for (KvrocksDataFormat format : values()) { + if (format.typeName.equals(typeName)) { + return format; + } + } + return null; + } + + public String getTypeName() { + return typeName; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataProcess.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataProcess.java new file mode 100644 index 00000000000..420574d4555 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataProcess.java @@ -0,0 +1,171 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.regex.Pattern; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.StreamEntry; +import redis.clients.jedis.StreamEntryID; + +import static java.util.Objects.requireNonNull; + +/** + * Reads data from a single Kvrocks key and converts it to relational rows. + * + *

The reading strategy depends on the Kvrocks data type of the key + * (detected via the {@code TYPE} command), and the parsing strategy depends + * on the configured {@link KvrocksDataFormat}. + */ +public class KvrocksDataProcess { + private final Jedis jedis; + private final String tableName; + private final String keyDelimiter; + private final KvrocksDataType dataType; + private final KvrocksDataFormat dataFormat; + private final List> fields; + private final ObjectMapper objectMapper; + + KvrocksDataProcess(Jedis jedis, KvrocksTableFieldInfo fieldInfo) { + this.jedis = jedis; + this.tableName = fieldInfo.getTableName(); + this.keyDelimiter = fieldInfo.getKeyDelimiter(); + this.fields = fieldInfo.getFields(); + + String type = jedis.type(tableName); + this.dataType = + requireNonNull(KvrocksDataType.fromTypeName(type)); + this.dataFormat = + requireNonNull( + KvrocksDataFormat.fromTypeName( + fieldInfo.getDataFormat())); + + this.objectMapper = new ObjectMapper(); + objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); + objectMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + objectMapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true); + } + + /** Reads all values for the key and returns them as relational rows. */ + public List read() { + switch (dataType) { + case STRING: + return readString(); + case LIST: + return parseValues(jedis.lrange(tableName, 0, -1)); + case SET: + return parseValues(jedis.smembers(tableName)); + case SORTED_SET: + return parseValues(jedis.zrange(tableName, 0, -1)); + case HASH: + return parseValues(jedis.hvals(tableName)); + case STREAM: + return readStream(); + default: + return new ArrayList<>(); + } + } + + /** STRING type: the key name itself is the lookup key; value is the data. */ + private List readString() { + List rows = new ArrayList<>(); + for (String key : jedis.keys(tableName)) { + String value = jedis.get(key); + if (value != null) { + rows.add(parseRow(value)); + } + } + return rows; + } + + /** Reads entries from a Kvrocks Stream via XRANGE. */ + private List readStream() { + List rows = new ArrayList<>(); + List entries = + jedis.xrange(tableName, (StreamEntryID) null, + (StreamEntryID) null, Integer.MAX_VALUE); + for (StreamEntry entry : entries) { + try { + String json = + objectMapper.writeValueAsString(entry.getFields()); + rows.add(parseRow(json)); + } catch (Exception e) { + throw new RuntimeException( + "Failed to serialize stream entry", e); + } + } + return rows; + } + + /** Parses a collection of string values into rows. */ + private List parseValues(Iterable values) { + List rows = new ArrayList<>(); + for (String value : values) { + rows.add(parseRow(value)); + } + return rows; + } + + /** Converts a single string value into a row array. */ + private Object[] parseRow(String value) { + switch (dataFormat) { + case RAW: + return new Object[]{value}; + case CSV: + return parseCsv(value); + case JSON: + return parseJson(value); + default: + throw new AssertionError("unexpected format: " + dataFormat); + } + } + + private Object[] parseCsv(String value) { + String[] parts = value.split(Pattern.quote(keyDelimiter)); + Object[] row = new Object[fields.size()]; + for (int i = 0; i < row.length; i++) { + row[i] = i < parts.length ? parts[i] : ""; + } + return row; + } + + private Object[] parseJson(String value) { + Object[] row = new Object[fields.size()]; + try { + JsonNode node = objectMapper.readTree(value); + for (int i = 0; i < row.length; i++) { + Object mapping = fields.get(i).get("mapping"); + if (mapping == null) { + row[i] = ""; + } else { + JsonNode found = node.findValue(mapping.toString()); + row[i] = found != null ? found : ""; + } + } + } catch (Exception e) { + throw new RuntimeException("Failed to parse JSON value: " + value, e); + } + return row; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataType.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataType.java new file mode 100644 index 00000000000..3bc6af2726f --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksDataType.java @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Data types supported by Kvrocks. + * + *

Kvrocks supports all classic Redis data types plus additional types + * such as JSON, Stream, and Bloom Filter. + */ +public enum KvrocksDataType { + + /** Basic key-value string. */ + STRING("string"), + + /** Field-value map. */ + HASH("hash"), + + /** Ordered list of strings. */ + LIST("list"), + + /** Unordered collection of unique strings. */ + SET("set"), + + /** Scored, sorted collection of unique strings. */ + SORTED_SET("zset"), + + /** Append-only log with consumer group support (since Kvrocks 2.2). */ + STREAM("stream"), + + /** Native JSON document storage (since Kvrocks 2.7). */ + JSON("ReJSON-RL"), + + /** Probabilistic membership filter (since Kvrocks 2.6). */ + BLOOM("MBbloom--"); + + private final String typeName; + + KvrocksDataType(String typeName) { + this.typeName = typeName; + } + + /** Resolves a Kvrocks {@code TYPE} command response to an enum value. */ + public static @Nullable KvrocksDataType fromTypeName(String typeName) { + for (KvrocksDataType type : values()) { + if (type.typeName.equals(typeName)) { + return type; + } + } + return null; + } + + public String getTypeName() { + return typeName; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksEnumerator.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksEnumerator.java new file mode 100644 index 00000000000..55850561ad9 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksEnumerator.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.Linq4j; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import redis.clients.jedis.Jedis; + +import static java.util.Objects.requireNonNull; + +/** + * Enumerator that reads rows from a single Kvrocks key. + * + *

On construction it borrows a {@link Jedis} connection from the + * schema-level {@link KvrocksJedisManager}, eagerly fetches all data + * via {@link KvrocksDataProcess}, then returns the connection to the + * pool. Iteration is over the materialised list. + */ +class KvrocksEnumerator implements Enumerator { + private final Enumerator enumerator; + + KvrocksEnumerator(KvrocksConfig config, KvrocksSchema schema, + String tableName) { + KvrocksTableFieldInfo fieldInfo = + schema.getTableFieldInfo(tableName); + KvrocksJedisManager manager = schema.getManager(); + + try (Jedis jedis = manager.getResource()) { + if (config.getPassword() != null + && !config.getPassword().isEmpty()) { + jedis.auth(config.getPassword()); + } + KvrocksDataProcess process = + new KvrocksDataProcess(jedis, fieldInfo); + List rows = process.read(); + enumerator = Linq4j.enumerator(rows); + } + } + + /** + * Derives the column name/type map from field metadata. + * + *

For {@link KvrocksDataFormat#RAW} a single {@code "key"} column is + * produced; for CSV and JSON the configured field list is used. + */ + static Map deduceRowType( + KvrocksTableFieldInfo fieldInfo) { + final Map columns = new LinkedHashMap<>(); + KvrocksDataFormat format = + requireNonNull(KvrocksDataFormat.fromTypeName(fieldInfo.getDataFormat())); + if (format == KvrocksDataFormat.RAW) { + columns.put("key", "key"); + } else { + for (LinkedHashMap field : fieldInfo.getFields()) { + columns.put(field.get("name").toString(), + field.get("type").toString()); + } + } + return columns; + } + + @Override public Object[] current() { + return enumerator.current(); + } + + @Override public boolean moveNext() { + return enumerator.moveNext(); + } + + @Override public void reset() { + enumerator.reset(); + } + + @Override public void close() { + enumerator.close(); + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksJedisManager.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksJedisManager.java new file mode 100644 index 00000000000..1c953cdf6a4 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksJedisManager.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.util.trace.CalciteTrace; + +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; + +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.cache.RemovalListener; +import com.google.common.cache.RemovalNotification; + +import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.Protocol; + +import static java.util.Objects.requireNonNull; + +/** + * Manages pooled connections to a Kvrocks instance via Jedis. + * + *

Because Kvrocks speaks the Redis wire protocol, the standard Jedis + * client works without modification. Connections are pooled per host using + * a Guava {@link LoadingCache}. + */ +public class KvrocksJedisManager implements AutoCloseable { + private static final Logger LOGGER = CalciteTrace.getPlannerTracer(); + + private final LoadingCache poolCache; + private final JedisPoolConfig poolConfig; + private final String host; + private final int port; + private final int database; + private final @Nullable String password; + + KvrocksJedisManager(String host, int port, int database, + @Nullable String password) { + this.host = host; + this.port = port; + this.database = database; + this.password = password; + + this.poolConfig = new JedisPoolConfig(); + poolConfig.setMaxTotal(GenericObjectPoolConfig.DEFAULT_MAX_TOTAL); + poolConfig.setMaxIdle(GenericObjectPoolConfig.DEFAULT_MAX_IDLE); + poolConfig.setMinIdle(GenericObjectPoolConfig.DEFAULT_MIN_IDLE); + + this.poolCache = CacheBuilder.newBuilder() + .removalListener(new PoolRemovalListener()) + .build(CacheLoader.from(this::createPool)); + } + + /** Returns a pooled {@link JedisPool} for the configured host. */ + public JedisPool getJedisPool() { + return poolCache.getUnchecked(host); + } + + /** Borrows a single {@link Jedis} connection from the pool. */ + public Jedis getResource() { + return getJedisPool().getResource(); + } + + private JedisPool createPool() { + String pwd = (password == null || password.isEmpty()) ? null : password; + return new JedisPool(poolConfig, host, port, + Protocol.DEFAULT_TIMEOUT, pwd, database); + } + + @Override public void close() { + poolCache.invalidateAll(); + } + + /** Destroys evicted pools so connections are released. */ + private static class PoolRemovalListener + implements RemovalListener { + @Override public void onRemoval( + RemovalNotification notification) { + try { + requireNonNull(notification.getValue()).destroy(); + } catch (Exception e) { + LOGGER.warn("Error destroying JedisPool for {}", notification.getKey(), e); + } + } + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchema.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchema.java new file mode 100644 index 00000000000..48bb2fe3458 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchema.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.model.JsonCustomTable; +import org.apache.calcite.schema.Table; +import org.apache.calcite.schema.impl.AbstractSchema; + +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Maps; + +import org.checkerframework.checker.nullness.qual.Nullable; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import static java.util.Objects.requireNonNull; + +/** + * Schema that exposes Kvrocks keys as relational tables. + * + *

Each entry in the {@code tables} list of the model JSON becomes a + * {@link KvrocksTable}. The schema holds connection parameters shared by + * all tables and owns a single {@link KvrocksJedisManager} that is reused + * across all scans. + */ +class KvrocksSchema extends AbstractSchema { + private static final String DATA_FORMAT = "dataFormat"; + private static final String FIELDS = "fields"; + private static final String KEY_DELIMITER = "keyDelimiter"; + private static final String OPERAND = "operand"; + + final String host; + final int port; + final int database; + final @Nullable String password; + final List> tables; + private final KvrocksJedisManager manager; + + KvrocksSchema(String host, int port, int database, + @Nullable String password, + List> tables) { + this.host = host; + this.port = port; + this.database = database; + this.password = password; + this.tables = tables; + this.manager = + new KvrocksJedisManager(host, port, database, password); + } + + /** Returns the shared connection manager for this schema. */ + KvrocksJedisManager getManager() { + return manager; + } + + @Override protected Map getTableMap() { + JsonCustomTable[] array = new JsonCustomTable[tables.size()]; + Set tableNames = Arrays.stream(tables.toArray(array)) + .map(t -> t.name) + .collect(Collectors.toSet()); + return Maps.asMap(ImmutableSet.copyOf(tableNames), + CacheBuilder.newBuilder() + .build(CacheLoader.from(this::table))); + } + + private Table table(String tableName) { + KvrocksConfig config = + new KvrocksConfig(host, port, database, password); + return KvrocksTable.create(KvrocksSchema.this, tableName, + config, null); + } + + /** + * Extracts field metadata for the named table from the model definition. + * + * @throws RuntimeException if dataFormat is missing/invalid or fields + * are empty + */ + @SuppressWarnings({"unchecked", "rawtypes"}) + KvrocksTableFieldInfo getTableFieldInfo(String tableName) { + KvrocksTableFieldInfo info = new KvrocksTableFieldInfo(); + List> fields = new ArrayList<>(); + String dataFormat = ""; + String keyDelimiter = ""; + + List jsonTables = + (List) (List) this.tables; + for (JsonCustomTable jsonTable : jsonTables) { + if (jsonTable.name.equals(tableName)) { + Map operand = + requireNonNull(jsonTable.operand, OPERAND); + + Object rawFormat = operand.get(DATA_FORMAT); + if (rawFormat == null || rawFormat.toString().isEmpty()) { + throw new RuntimeException( + "dataFormat is invalid, it must be raw, csv or json"); + } + KvrocksDataFormat fmt = + KvrocksDataFormat.fromTypeName(rawFormat.toString()); + if (fmt == null) { + throw new RuntimeException( + "dataFormat is invalid, it must be raw, csv or json"); + } + Object rawFields = operand.get(FIELDS); + if (rawFields == null + || (rawFields instanceof List + && ((List) rawFields).isEmpty())) { + throw new RuntimeException("fields is null"); + } + + dataFormat = operand.get(DATA_FORMAT).toString(); + fields = + (List>) operand.get(FIELDS); + if (operand.get(KEY_DELIMITER) != null) { + keyDelimiter = operand.get(KEY_DELIMITER).toString(); + } + break; + } + } + + info.setTableName(tableName); + info.setDataFormat(dataFormat); + info.setFields(fields); + if (keyDelimiter != null && !keyDelimiter.isEmpty()) { + info.setKeyDelimiter(keyDelimiter); + } + return info; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchemaFactory.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchemaFactory.java new file mode 100644 index 00000000000..576c35610d7 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksSchemaFactory.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.schema.Schema; +import org.apache.calcite.schema.SchemaFactory; +import org.apache.calcite.schema.SchemaPlus; + +import java.util.List; +import java.util.Map; + +import static com.google.common.base.Preconditions.checkArgument; + +import static java.lang.Integer.parseInt; + +/** + * Factory that creates a {@link KvrocksSchema}. + * + *

Reads connection parameters and table definitions from the model JSON. + * + *

Required operand keys: {@code host}, {@code port}, {@code database}, + * {@code tables}. Optional: {@code password}. + */ +@SuppressWarnings("UnusedDeclaration") +public class KvrocksSchemaFactory implements SchemaFactory { + + public KvrocksSchemaFactory() { + } + + @Override public Schema create(SchemaPlus parentSchema, String name, + Map operand) { + checkArgument(operand.get("host") != null, "host must be specified"); + checkArgument(operand.get("port") != null, "port must be specified"); + checkArgument(operand.get("database") != null, + "database must be specified"); + checkArgument(operand.get("tables") != null, + "tables must be specified"); + + String host = operand.get("host").toString(); + int port = (int) operand.get("port"); + int database = parseInt(operand.get("database").toString()); + + String password = operand.get("password") == null + ? null : operand.get("password").toString(); + + @SuppressWarnings("unchecked") + List> tables = (List) operand.get("tables"); + + return new KvrocksSchema(host, port, database, password, tables); + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTable.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTable.java new file mode 100644 index 00000000000..f3979159e3f --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTable.java @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.DataContext; +import org.apache.calcite.linq4j.AbstractEnumerable; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelProtoDataType; +import org.apache.calcite.schema.ScannableTable; +import org.apache.calcite.schema.Table; +import org.apache.calcite.schema.impl.AbstractTable; +import org.apache.calcite.util.Pair; + +import com.google.common.collect.ImmutableMap; + +import org.checkerframework.checker.nullness.qual.Nullable; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * A Calcite {@link Table} backed by a single Kvrocks key. + * + *

Implements {@link ScannableTable}: every SQL query triggers a full + * read of the underlying Kvrocks data structure. + */ +public class KvrocksTable extends AbstractTable implements ScannableTable { + + final KvrocksSchema schema; + final String tableName; + final @Nullable RelProtoDataType protoRowType; + final ImmutableMap allFields; + final String dataFormat; + final KvrocksConfig kvrocksConfig; + + KvrocksTable(KvrocksSchema schema, String tableName, + @Nullable RelProtoDataType protoRowType, + Map allFields, String dataFormat, + KvrocksConfig kvrocksConfig) { + this.schema = schema; + this.tableName = tableName; + this.protoRowType = protoRowType; + this.allFields = allFields == null + ? ImmutableMap.of() : ImmutableMap.copyOf(allFields); + this.dataFormat = dataFormat; + this.kvrocksConfig = kvrocksConfig; + } + + @Override public RelDataType getRowType(RelDataTypeFactory typeFactory) { + if (protoRowType != null) { + return protoRowType.apply(typeFactory); + } + final List names = new ArrayList<>(allFields.size()); + final List types = new ArrayList<>(allFields.size()); + for (Map.Entry entry : allFields.entrySet()) { + names.add(entry.getKey()); + types.add(typeFactory.createJavaType(entry.getValue().getClass())); + } + return typeFactory.createStructType(Pair.zip(names, types)); + } + + @Override public Enumerable<@Nullable Object[]> scan(DataContext root) { + return new AbstractEnumerable() { + @Override public Enumerator enumerator() { + return new KvrocksEnumerator(kvrocksConfig, schema, tableName); + } + }; + } + + /** Creates a table from schema-level metadata. */ + static Table create(KvrocksSchema schema, String tableName, + KvrocksConfig config, @Nullable RelProtoDataType protoRowType) { + KvrocksTableFieldInfo fieldInfo = schema.getTableFieldInfo(tableName); + Map allFields = + KvrocksEnumerator.deduceRowType(fieldInfo); + return new KvrocksTable(schema, tableName, protoRowType, + allFields, fieldInfo.getDataFormat(), config); + } + + /** Creates a table from an operand map (used by {@link KvrocksTableFactory}). */ + static Table create(KvrocksSchema schema, String tableName, + Map operand, @Nullable RelProtoDataType protoRowType) { + KvrocksConfig config = + new KvrocksConfig(schema.host, schema.port, + schema.database, schema.password); + return create(schema, tableName, config, protoRowType); + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFactory.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFactory.java new file mode 100644 index 00000000000..99bc4de5d8b --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFactory.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeImpl; +import org.apache.calcite.rel.type.RelProtoDataType; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.Table; +import org.apache.calcite.schema.TableFactory; + +import org.checkerframework.checker.nullness.qual.Nullable; + +import java.util.Map; + +/** + * Implementation of {@link TableFactory} for Kvrocks. + * + *

Called by Calcite when a table is defined in the model JSON with + * {@code "factory": "...KvrocksTableFactory"}. + */ +public class KvrocksTableFactory implements TableFactory { + + @SuppressWarnings("unused") + public static final KvrocksTableFactory INSTANCE = new KvrocksTableFactory(); + + private KvrocksTableFactory() { + } + + @Override public Table create(SchemaPlus schema, String tableName, + Map operand, @Nullable RelDataType rowType) { + final KvrocksSchema kvrocksSchema = schema.unwrap(KvrocksSchema.class); + final RelProtoDataType protoRowType = + rowType != null ? RelDataTypeImpl.proto(rowType) : null; + return KvrocksTable.create(kvrocksSchema, tableName, operand, protoRowType); + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFieldInfo.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFieldInfo.java new file mode 100644 index 00000000000..676407b808b --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/KvrocksTableFieldInfo.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import java.util.LinkedHashMap; +import java.util.List; + +/** + * Metadata describing how a Kvrocks key maps to a relational table. + * + *

Carries the key name (used as Kvrocks key), the data format that + * determines how stored values are parsed, the column definitions, and + * an optional delimiter for CSV parsing. + */ +public class KvrocksTableFieldInfo { + private String tableName; + private String dataFormat; + private List> fields; + private String keyDelimiter = ":"; + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getDataFormat() { + return dataFormat; + } + + public void setDataFormat(String dataFormat) { + this.dataFormat = dataFormat; + } + + public List> getFields() { + return fields; + } + + public void setFields(List> fields) { + this.fields = fields; + } + + public String getKeyDelimiter() { + return keyDelimiter; + } + + public void setKeyDelimiter(String keyDelimiter) { + this.keyDelimiter = keyDelimiter; + } +} diff --git a/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/package-info.java b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/package-info.java new file mode 100644 index 00000000000..c419bb9ece7 --- /dev/null +++ b/kvrocks/src/main/java/org/apache/calcite/adapter/kvrocks/package-info.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * Calcite query provider that reads from Apache Kvrocks. + * + *

Kvrocks is a distributed key-value NoSQL database that uses RocksDB as + * its storage engine and is compatible with the Redis protocol. This adapter + * maps Kvrocks data structures (String, Hash, List, Set, Sorted Set, JSON, + * Stream) onto relational tables that can be queried via SQL. + */ +@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.FIELD) +@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.PARAMETER) +@DefaultQualifier(value = NonNull.class, locations = TypeUseLocation.RETURN) +package org.apache.calcite.adapter.kvrocks; + +import org.checkerframework.checker.nullness.qual.NonNull; +import org.checkerframework.framework.qual.DefaultQualifier; +import org.checkerframework.framework.qual.TypeUseLocation; diff --git a/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterCaseBase.java b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterCaseBase.java new file mode 100644 index 00000000000..f9e88eb5098 --- /dev/null +++ b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterCaseBase.java @@ -0,0 +1,118 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.test.CalciteAssert; +import org.apache.calcite.util.Sources; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Integration tests for the Kvrocks adapter. + * + *

Validates SQL queries against data populated into a live Kvrocks + * instance via Docker. + */ +public class KvrocksAdapterCaseBase extends KvrocksDataCaseBase { + + private final String filePath = + Sources.of(KvrocksAdapterCaseBase.class.getResource("/kvrocks-mix-model.json")) + .file().getAbsolutePath(); + + private String model; + + @SuppressWarnings("unchecked") + private static final Map TABLE_ROW_COUNTS = new HashMap<>(15); + + static { + TABLE_ROW_COUNTS.put("raw_01", 1); + TABLE_ROW_COUNTS.put("raw_02", 2); + TABLE_ROW_COUNTS.put("raw_03", 2); + TABLE_ROW_COUNTS.put("raw_04", 2); + TABLE_ROW_COUNTS.put("raw_05", 2); + TABLE_ROW_COUNTS.put("csv_01", 1); + TABLE_ROW_COUNTS.put("csv_02", 2); + TABLE_ROW_COUNTS.put("csv_03", 2); + TABLE_ROW_COUNTS.put("csv_04", 2); + TABLE_ROW_COUNTS.put("csv_05", 2); + TABLE_ROW_COUNTS.put("json_01", 1); + TABLE_ROW_COUNTS.put("json_02", 2); + TABLE_ROW_COUNTS.put("json_03", 2); + TABLE_ROW_COUNTS.put("json_04", 2); + TABLE_ROW_COUNTS.put("json_05", 2); + } + + @BeforeEach + @Override public void makeData() { + super.makeData(); + readModelJson(); + } + + private CalciteAssert.AssertQuery sql(String sql) { + assertNotNull(model, "model cannot be null"); + return CalciteAssert.model(model) + .enable(isKvrocksRunning()) + .query(sql); + } + + private void readModelJson() { + try { + ObjectMapper mapper = new ObjectMapper(); + mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); + mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + mapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true); + + File file = new File(filePath); + if (file.exists()) { + JsonNode root = mapper.readTree(file); + model = root.toString() + .replace("6666", + Integer.toString(getKvrocksServerPort())); + } + } catch (Exception e) { + throw new RuntimeException("Failed to read model JSON", e); + } + } + + @Test void testKvrocksBySql() { + assumeTrue(isKvrocksRunning()); + TABLE_ROW_COUNTS.forEach((table, count) -> { + String query = "SELECT count(*) AS c FROM \"" + table + "\" WHERE true"; + sql(query).returnsUnordered("C=" + count); + }); + } + + @Test void testSqlWithJoin() { + assumeTrue(isKvrocksRunning()); + String query = "SELECT a.DEPTNO, b.NAME " + + "FROM \"csv_01\" a LEFT JOIN \"json_02\" b " + + "ON a.DEPTNO = b.DEPTNO WHERE true"; + sql(query).returnsUnordered("DEPTNO=10; NAME=\"Sales1\""); + } +} diff --git a/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterConfigCaseBase.java b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterConfigCaseBase.java new file mode 100644 index 00000000000..c857241e2ae --- /dev/null +++ b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksAdapterConfigCaseBase.java @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.test.CalciteAssert; + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Tests that invalid model configurations are rejected with clear errors. + */ +public class KvrocksAdapterConfigCaseBase extends KvrocksAdapterCaseBase { + + private String configModel; + + private void readModelFromJsonString(String jsonString) { + try { + ObjectMapper mapper = new ObjectMapper(); + mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); + mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + mapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true); + JsonNode root = mapper.readTree(jsonString); + configModel = root.toString() + .replace("6666", + Integer.toString(getKvrocksServerPort())); + } catch (Exception e) { + throw new RuntimeException("Failed to read model from JSON string", e); + } + } + + @Test void testDataFormatEmptyException() { + assumeTrue(isKvrocksRunning()); + String json = "{" + + " \"version\": \"1.0\"," + + " \"defaultSchema\": \"kvrocks\"," + + " \"schemas\": [{" + + " \"type\": \"custom\"," + + " \"name\": \"foodmart\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksSchemaFactory\"," + + " \"operand\": {" + + " \"host\": \"localhost\"," + + " \"port\": 6666," + + " \"database\": 0," + + " \"password\": \"\"" + + " }," + + " \"tables\": [{" + + " \"name\": \"csv_05\"," + + " \"type\": \"custom\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksTableFactory\"," + + " \"operand\": {" + + " \"dataFormat\": \"\"," + + " \"keyDelimiter\": \":\"," + + " \"fields\": [" + + " {\"name\": \"DEPTNO\", \"type\": \"varchar\", \"mapping\": 0}," + + " {\"name\": \"NAME\", \"type\": \"varchar\", \"mapping\": 1}" + + " ]" + + " }" + + " }]" + + " }]" + + "}"; + readModelFromJsonString(json); + assertNotNull(configModel, "model cannot be null"); + CalciteAssert.model(configModel) + .enable(isKvrocksRunning()) + .connectThrows("dataFormat is invalid, it must be raw, csv or json"); + } + + @Test void testDataFieldsEmptyException() { + assumeTrue(isKvrocksRunning()); + String json = "{" + + " \"version\": \"1.0\"," + + " \"defaultSchema\": \"kvrocks\"," + + " \"schemas\": [{" + + " \"type\": \"custom\"," + + " \"name\": \"foodmart\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksSchemaFactory\"," + + " \"operand\": {" + + " \"host\": \"localhost\"," + + " \"port\": 6666," + + " \"database\": 0," + + " \"password\": \"\"" + + " }," + + " \"tables\": [{" + + " \"name\": \"csv_05\"," + + " \"type\": \"custom\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksTableFactory\"," + + " \"operand\": {" + + " \"dataFormat\": \"csv\"," + + " \"keyDelimiter\": \":\"," + + " \"fields\": []" + + " }" + + " }]" + + " }]" + + "}"; + readModelFromJsonString(json); + assertNotNull(configModel, "model cannot be null"); + CalciteAssert.model(configModel) + .enable(isKvrocksRunning()) + .connectThrows("fields is null"); + } + + @Test void testDataFormatInvalidException() { + assumeTrue(isKvrocksRunning()); + String json = "{" + + " \"version\": \"1.0\"," + + " \"defaultSchema\": \"kvrocks\"," + + " \"schemas\": [{" + + " \"type\": \"custom\"," + + " \"name\": \"foodmart\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksSchemaFactory\"," + + " \"operand\": {" + + " \"host\": \"localhost\"," + + " \"port\": 6666," + + " \"database\": 0," + + " \"password\": \"\"" + + " }," + + " \"tables\": [{" + + " \"name\": \"csv_05\"," + + " \"type\": \"custom\"," + + " \"factory\": \"org.apache.calcite.adapter.kvrocks.KvrocksTableFactory\"," + + " \"operand\": {" + + " \"dataFormat\": \"CSV\"," + + " \"keyDelimiter\": \":\"," + + " \"fields\": [" + + " {\"name\": \"DEPTNO\", \"type\": \"varchar\", \"mapping\": 0}," + + " {\"name\": \"NAME\", \"type\": \"varchar\", \"mapping\": 1}" + + " ]" + + " }" + + " }]" + + " }]" + + "}"; + readModelFromJsonString(json); + assertNotNull(configModel, "model cannot be null"); + CalciteAssert.model(configModel) + .enable(isKvrocksRunning()) + .connectThrows("dataFormat is invalid, it must be raw, csv or json"); + } +} diff --git a/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksCaseBase.java b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksCaseBase.java new file mode 100644 index 00000000000..756eae795a6 --- /dev/null +++ b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksCaseBase.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.apache.calcite.config.CalciteSystemProperty; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.GenericContainer; + +import java.io.IOException; +import java.net.ServerSocket; +import java.util.logging.Logger; + +/** + * Base class for Kvrocks adapter tests. + * + *

Uses a Kvrocks Docker container ({@code apache/kvrocks:latest}) when + * Docker is available; otherwise tests are skipped via + * {@link #isKvrocksRunning()}. + */ +@Execution(ExecutionMode.SAME_THREAD) +public abstract class KvrocksCaseBase { + + private static final Logger LOGGER = Logger.getLogger(KvrocksCaseBase.class.getName()); + + /** Default Kvrocks port inside the container. */ + private static final int KVROCKS_PORT = 6666; + + private static final GenericContainer KVROCKS_CONTAINER = + new GenericContainer<>("apache/kvrocks:latest") + .withExposedPorts(KVROCKS_PORT); + + private static boolean containerStarted = false; + + @BeforeAll + public static void startContainer() { + if (CalciteSystemProperty.TEST_WITH_DOCKER_CONTAINER.value() + && DockerClientFactory.instance().isDockerAvailable()) { + KVROCKS_CONTAINER.start(); + containerStarted = true; + LOGGER.info("Kvrocks container started on port " + getKvrocksServerPort()); + } else { + LOGGER.warning("Docker is not available; Kvrocks tests will be skipped"); + } + } + + @AfterAll + public static void stopContainer() { + if (containerStarted && KVROCKS_CONTAINER.isRunning()) { + KVROCKS_CONTAINER.stop(); + } + } + + /** Whether the Kvrocks server is reachable. */ + static boolean isKvrocksRunning() { + return containerStarted && KVROCKS_CONTAINER.isRunning(); + } + + static int getKvrocksServerPort() { + return KVROCKS_CONTAINER.getMappedPort(KVROCKS_PORT); + } + + static String getKvrocksServerHost() { + return KVROCKS_CONTAINER.getHost(); + } + + /** + * Finds a free port on the host. + */ + static int getAvailablePort() { + for (int i = 0; i < 50; i++) { + try (ServerSocket socket = new ServerSocket(0)) { + int port = socket.getLocalPort(); + if (port != 0) { + return port; + } + } catch (IOException ignored) { + } + } + throw new RuntimeException("Could not find an available port"); + } +} diff --git a/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksDataCaseBase.java b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksDataCaseBase.java new file mode 100644 index 00000000000..6f2ee3dcc68 --- /dev/null +++ b/kvrocks/src/test/java/org/apache/calcite/adapter/kvrocks/KvrocksDataCaseBase.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.calcite.adapter.kvrocks; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; + +import java.util.HashMap; +import java.util.Map; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisPool; +import redis.clients.jedis.JedisPoolConfig; + +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Base class that populates test data into Kvrocks before each test. + * + *

Creates 15 tables across 3 data formats (raw, csv, json) x 5 data + * types (string, list, set, sorted-set, hash). + */ +public class KvrocksDataCaseBase extends KvrocksCaseBase { + private JedisPool pool; + + final String[] tableNames = { + "raw_01", "raw_02", "raw_03", "raw_04", "raw_05", + "csv_01", "csv_02", "csv_03", "csv_04", "csv_05", + "json_01", "json_02", "json_03", "json_04", "json_05" + }; + + @BeforeEach + public void setUp() { + assumeTrue(isKvrocksRunning(), "Kvrocks container is not running"); + + JedisPoolConfig config = new JedisPoolConfig(); + config.setMaxTotal(10); + pool = new JedisPool(config, getKvrocksServerHost(), getKvrocksServerPort()); + + try (Jedis jedis = pool.getResource()) { + jedis.flushAll(); + } + } + + /** Populates all 15 test tables into Kvrocks. */ + public void makeData() { + try (Jedis jedis = pool.getResource()) { + jedis.del(tableNames); + + // STRING type + jedis.set("raw_01", "123"); + jedis.set("json_01", "{\"DEPTNO\":10,\"NAME\":\"Sales\"}"); + jedis.set("csv_01", "10:Sales"); + + // LIST type + jedis.lpush("raw_02", "book1", "book2"); + jedis.lpush("json_02", + "{\"DEPTNO\":10,\"NAME\":\"Sales1\"}", + "{\"DEPTNO\":20,\"NAME\":\"Sales2\"}"); + jedis.lpush("csv_02", "10:Sales", "20:Sales"); + + // SET type + jedis.sadd("raw_03", "user1", "user2"); + jedis.sadd("json_03", + "{\"DEPTNO\":10,\"NAME\":\"Sales1\"}", + "{\"DEPTNO\":20,\"NAME\":\"Sales1\"}"); + jedis.sadd("csv_03", "10:Sales", "20:Sales"); + + // SORTED SET type + jedis.zadd("raw_04", 22, "user3"); + jedis.zadd("raw_04", 24, "user4"); + jedis.zadd("json_04", 1, "{\"DEPTNO\":10,\"NAME\":\"Sales1\"}"); + jedis.zadd("json_04", 2, "{\"DEPTNO\":11,\"NAME\":\"Sales2\"}"); + jedis.zadd("csv_04", 1, "10:Sales"); + jedis.zadd("csv_04", 2, "20:Sales"); + + // HASH type + Map rawHash = new HashMap<>(); + rawHash.put("stuA", "a1"); + rawHash.put("stuB", "b2"); + jedis.hmset("raw_05", rawHash); + + Map jsonHash = new HashMap<>(); + jsonHash.put("stuA", "{\"DEPTNO\":10,\"NAME\":\"stuA\"}"); + jsonHash.put("stuB", "{\"DEPTNO\":10,\"NAME\":\"stuB\"}"); + jedis.hmset("json_05", jsonHash); + + Map csvHash = new HashMap<>(); + csvHash.put("stuA", "10:Sales"); + csvHash.put("stuB", "20:Sales"); + jedis.hmset("csv_05", csvHash); + } + } + + @AfterEach + public void tearDown() { + if (pool != null) { + pool.destroy(); + } + } +} diff --git a/kvrocks/src/test/resources/kvrocks-mix-model.json b/kvrocks/src/test/resources/kvrocks-mix-model.json new file mode 100644 index 00000000000..f59f24bf905 --- /dev/null +++ b/kvrocks/src/test/resources/kvrocks-mix-model.json @@ -0,0 +1,200 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +{ + "version": "1.0", + "defaultSchema": "foodmart", + "schemas": [ + { + "type": "custom", + "name": "foodmart", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksSchemaFactory", + "operand": { + "host": "localhost", + "port": 6666, + "database": 0, + "password": "" + }, + "tables": [ + { + "name": "csv_01", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "csv", + "keyDelimiter": ":", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": 0 }, + { "name": "NAME", "type": "varchar", "mapping": 1 } + ] + } + }, + { + "name": "csv_02", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "csv", + "keyDelimiter": ":", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": 0 }, + { "name": "NAME", "type": "varchar", "mapping": 1 } + ] + } + }, + { + "name": "csv_03", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "csv", + "keyDelimiter": ":", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": 0 }, + { "name": "NAME", "type": "varchar", "mapping": 1 } + ] + } + }, + { + "name": "csv_04", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "csv", + "keyDelimiter": ":", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": 0 }, + { "name": "NAME", "type": "varchar", "mapping": 1 } + ] + } + }, + { + "name": "csv_05", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "csv", + "keyDelimiter": ":", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": 0 }, + { "name": "NAME", "type": "varchar", "mapping": 1 } + ] + } + }, + { + "name": "json_01", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "json", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": "DEPTNO" }, + { "name": "NAME", "type": "varchar", "mapping": "NAME" } + ] + } + }, + { + "name": "json_02", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "json", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": "DEPTNO" }, + { "name": "NAME", "type": "varchar", "mapping": "NAME" } + ] + } + }, + { + "name": "json_03", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "json", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": "DEPTNO" }, + { "name": "NAME", "type": "varchar", "mapping": "NAME" } + ] + } + }, + { + "name": "json_04", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "json", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": "DEPTNO" }, + { "name": "NAME", "type": "varchar", "mapping": "NAME" } + ] + } + }, + { + "name": "json_05", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "json", + "fields": [ + { "name": "DEPTNO", "type": "varchar", "mapping": "DEPTNO" }, + { "name": "NAME", "type": "varchar", "mapping": "NAME" } + ] + } + }, + { + "name": "raw_01", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "raw", + "fields": [ + { "name": "key", "type": "varchar", "mapping": "key" } + ] + } + }, + { + "name": "raw_02", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "raw", + "fields": [ + { "name": "key", "type": "varchar", "mapping": "key" } + ] + } + }, + { + "name": "raw_03", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "raw", + "fields": [ + { "name": "key", "type": "varchar", "mapping": "key" } + ] + } + }, + { + "name": "raw_04", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "raw", + "fields": [ + { "name": "key", "type": "varchar", "mapping": "key" } + ] + } + }, + { + "name": "raw_05", + "factory": "org.apache.calcite.adapter.kvrocks.KvrocksTableFactory", + "operand": { + "dataFormat": "raw", + "fields": [ + { "name": "key", "type": "varchar", "mapping": "key" } + ] + } + } + ] + } + ] +} diff --git a/kvrocks/src/test/resources/log4j2-test.xml b/kvrocks/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000000..4f129b80dd8 --- /dev/null +++ b/kvrocks/src/test/resources/log4j2-test.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + diff --git a/settings.gradle.kts b/settings.gradle.kts index 67695f630c5..22a5942e72f 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -75,6 +75,7 @@ include( "geode", "innodb", "kafka", + "kvrocks", "linq4j", "mongodb", "pig",