feat: add redis connector

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/pom.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>geaflow-dsl-connector</artifactId>
+        <groupId>org.apache.geaflow</groupId>
+        <version>0.6.8-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>geaflow-dsl-connector-redis</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-common</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.geaflow</groupId>
+            <artifactId>geaflow-dsl-connector-api</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+            <version>3.7.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.github.fppt</groupId>
+            <artifactId>jedis-mock</artifactId>
+            <version>1.0.6</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisConfigKeys.java

@@ -0,0 +1,45 @@
+/*
+ * 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.geaflow.dsl.connector.redis;
+
+import org.apache.geaflow.common.config.ConfigKey;
+import org.apache.geaflow.common.config.ConfigKeys;
+
+public class RedisConfigKeys {
+    public static final ConfigKey GEAFLOW_DSL_REDIS_HOST = ConfigKeys
+        .key("geaflow.dsl.redis.host")
+        .defaultValue("localhost")
+        .description("Redis host.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_PORT = ConfigKeys
+        .key("geaflow.dsl.redis.port")
+        .defaultValue(6379)
+        .description("Redis port.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_PASSWORD = ConfigKeys
+        .key("geaflow.dsl.redis.password")
+        .noDefaultValue()
+        .description("Redis password.");
+
+    public static final ConfigKey GEAFLOW_DSL_REDIS_DB_INDEX = ConfigKeys
+        .key("geaflow.dsl.redis.db.index")
+        .defaultValue(0)
+        .description("Redis database index.");
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableConnector.java

@@ -0,0 +1,37 @@
+/*
+ * 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.geaflow.dsl.connector.redis;
+
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.dsl.connector.api.TableSink;
+import org.apache.geaflow.dsl.connector.api.TableWritableConnector;
+
+public class RedisTableConnector implements TableWritableConnector {
+
+    @Override
+    public String getType() {
+        return "REDIS";
+    }
+
+    @Override
+    public TableSink createSink(Configuration conf) {
+        return new RedisTableSink();
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/java/org/apache/geaflow/dsl/connector/redis/RedisTableSink.java

@@ -0,0 +1,99 @@
+/*
+ * 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.geaflow.dsl.connector.redis;
+
+import java.io.IOException;
+import org.apache.geaflow.api.context.RuntimeContext;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.connector.api.TableSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+
+public class RedisTableSink implements TableSink {
+    private static final Logger LOGGER = LoggerFactory.getLogger(RedisTableSink.class);
+
+    private Configuration tableConf;
+    private StructType schema;
+    private transient JedisPool jedisPool;
+
+    @Override
+    public void init(Configuration tableConf, StructType schema) {
+        this.tableConf = tableConf;
+        this.schema = schema;
+    }
+
+    @Override
+    public void open(RuntimeContext context) {
+        String host = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_HOST);
+        int port = tableConf.getInteger(RedisConfigKeys.GEAFLOW_DSL_REDIS_PORT);
+        String password = tableConf.getString(RedisConfigKeys.GEAFLOW_DSL_REDIS_PASSWORD, (String) null);
+        int dbIndex = tableConf.getInteger(RedisConfigKeys.GEAFLOW_DSL_REDIS_DB_INDEX);
+
+        JedisPoolConfig poolConfig = new JedisPoolConfig();
+        poolConfig.setMaxTotal(10);
+
+        if (password != null && !password.isEmpty()) {
+            this.jedisPool = new JedisPool(poolConfig, host, port, 2000, password, dbIndex);
+        } else {
+            this.jedisPool = new JedisPool(poolConfig, host, port, 2000, null, dbIndex);
+        }
+        LOGGER.info("Redis sink opened. Host: {}, Port: {}", host, port);
+    }
+
+    @Override
+    public void write(Row row) throws IOException {
+        if (row == null || schema.getFields().size() < 2) {
+            return;
+        }
+        // Assume first column is key
+        String key = String.valueOf(row.getField(0, schema.getType(0)));
+
+        try (Jedis jedis = jedisPool.getResource()) {
+            // Write other columns as hash fields
+            for (int i = 1; i < schema.getFields().size(); i++) {
+                String field = schema.getField(i).getName();
+                Object value = row.getField(i, schema.getType(i));
+                if (value != null) {
+                    jedis.hset(key, field, String.valueOf(value));
+                }
+            }
+        } catch (Exception e) {
+            throw new IOException("Failed to write to Redis", e);
+        }
+    }
+
+    @Override
+    public void finish() throws IOException {
+        // No-op for Redis sink as operations are immediate
+    }
+
+    @Override
+    public void close() {
+        if (jedisPool != null) {
+            jedisPool.close();
+        }
+        LOGGER.info("Redis sink closed.");
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/main/resources/META-INF/services/org.apache.geaflow.dsl.connector.api.TableConnector

@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.apache.geaflow.dsl.connector.redis.RedisTableConnector

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-redis/src/test/java/org/apache/geaflow/dsl/connector/redis/RedisTableSinkTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.geaflow.dsl.connector.redis;
+
+import com.github.fppt.jedismock.RedisServer;
+import java.io.IOException;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.common.type.Types;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import redis.clients.jedis.Jedis;
+
+public class RedisTableSinkTest {
+
+    private RedisServer server;
+    private RedisTableSink sink;
+
+    @BeforeMethod
+    public void setUp() throws IOException {
+        server = RedisServer.newRedisServer();
+        server.start();
+
+        sink = new RedisTableSink();
+        Configuration conf = new Configuration();
+        conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_HOST, server.getHost());
+        conf.put(RedisConfigKeys.GEAFLOW_DSL_REDIS_PORT, String.valueOf(server.getBindPort()));
+
+        StructType schema = new StructType(
+            new TableField("id", Types.INTEGER, false),
+            new TableField("name", Types.STRING, false),
+            new TableField("age", Types.INTEGER, false)
+        );
+
+        sink.init(conf, schema);
+        sink.open(null);
+    }
+
+    @AfterMethod
+    public void tearDown() throws IOException {
+        if (sink != null) {
+            sink.close();
+        }
+        if (server != null) {
+            server.stop();
+        }
+    }
+
+    @Test
+    public void testWrite() throws IOException {
+        Row row = ObjectRow.create(1, "alice", 20);
+        sink.write(row);
+
+        try (Jedis jedis = new Jedis(server.getHost(), server.getBindPort())) {
+            String name = jedis.hget("1", "name");
+            String age = jedis.hget("1", "age");
+
+            Assert.assertEquals(name, "alice");
+            Assert.assertEquals(age, "20");
+        }
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml

@@ -48,6 +48,7 @@
         <module>geaflow-dsl-connector-random</module>
         <module>geaflow-dsl-connector-paimon</module>
         <module>geaflow-dsl-connector-neo4j</module>
+        <module>geaflow-dsl-connector-redis</module>
         <module>geaflow-dsl-connector-elasticsearch</module>
     </modules>