This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 3f88dc9719 [Feature][Transform-V2] Add Base64 SQL functions (#11114)
3f88dc9719 is described below
commit 3f88dc9719622c669ed05a3719c02775628addb4
Author: Kang Myeong Gwan <[email protected]>
AuthorDate: Sat Jun 20 07:06:10 2026 +0900
[Feature][Transform-V2] Add Base64 SQL functions (#11114)
---
docs/en/transforms/sql-functions.md | 37 ++++++-
docs/zh/transforms/sql-functions.md | 38 ++++++-
.../test/resources/sql_transform/func_string.conf | 32 +++++-
.../transform/sql/zeta/ZetaSQLFunction.java | 6 ++
.../seatunnel/transform/sql/zeta/ZetaSQLType.java | 2 +
.../sql/zeta/functions/StringFunction.java | 43 ++++++++
.../transform/sql/SQLStringFunctionsTest.java | 75 ++++++++++++++
.../sql/zeta/functions/StringFunctionTest.java | 111 +++++++++++++++++++++
8 files changed, 340 insertions(+), 4 deletions(-)
diff --git a/docs/en/transforms/sql-functions.md
b/docs/en/transforms/sql-functions.md
index 2c6bd6bc17..28b8372067 100644
--- a/docs/en/transforms/sql-functions.md
+++ b/docs/en/transforms/sql-functions.md
@@ -94,6 +94,42 @@ Example:
RAWTOHEX(DATA)
+### TO_BASE64
+
+```TO_BASE64(value[, charset]) -> STRING```
+
+Encodes a string or bytes to Base64.
+
+The default charset is `UTF-8`. You can specify another charset.
+
+For bytes input, the charset argument is not supported because the value is
already raw bytes.
+
+Returns **NULL** if value is **NULL**.
+
+Example:
+
+TO_BASE64(NAME)
+
+TO_BASE64(NAME, 'UTF-16')
+
+TO_BASE64(BINARY_PAYLOAD)
+
+### FROM_BASE64
+
+```FROM_BASE64(value[, charset]) -> STRING```
+
+Decodes a Base64 string to text.
+
+The default charset is `UTF-8`. You can specify another charset.
+
+Returns **NULL** if value is **NULL**.
+
+Example:
+
+FROM_BASE64(ENCODED_NAME)
+
+FROM_BASE64(TO_BASE64(NAME, 'UTF-16'), 'UTF-16')
+
### INSERT
```INSERT(originalString, startInt, lengthInt, addString) -> STRING```
@@ -1350,4 +1386,3 @@ Normalizes a vector to unit length (magnitude = 1). This
is useful for computing
```sql
SELECT id, VECTOR_NORMALIZE(embedding) as normalized_embedding FROM table
```
-
diff --git a/docs/zh/transforms/sql-functions.md
b/docs/zh/transforms/sql-functions.md
index 5a2536fa42..9349ee1291 100644
--- a/docs/zh/transforms/sql-functions.md
+++ b/docs/zh/transforms/sql-functions.md
@@ -94,6 +94,42 @@ HEXTORAW(DATA)
RAWTOHEX(DATA)
+### TO_BASE64
+
+```TO_BASE64(value[, charset]) -> STRING```
+
+将字符串或字节编码为 Base64。
+
+默认字符集为 `UTF-8`。可以指定其他字符集。
+
+对于字节输入，不支持 charset 参数，因为该值已经是原始字节。
+
+如果 value 为 **NULL**，返回 **NULL**。
+
+示例:
+
+TO_BASE64(NAME)
+
+TO_BASE64(NAME, 'UTF-16')
+
+TO_BASE64(BINARY_PAYLOAD)
+
+### FROM_BASE64
+
+```FROM_BASE64(value[, charset]) -> STRING```
+
+将 Base64 字符串解码为文本。
+
+默认字符集为 `UTF-8`。可以指定其他字符集。
+
+如果 value 为 **NULL**，返回 **NULL**。
+
+示例:
+
+FROM_BASE64(ENCODED_NAME)
+
+FROM_BASE64(TO_BASE64(NAME, 'UTF-16'), 'UTF-16')
+
### INSERT
```INSERT(originalString, startInt, lengthInt, addString) -> STRING```
@@ -1353,4 +1389,4 @@ SELECT id, VECTOR_REDUCE(embedding, 64,
'SPARSE_RANDOM_PROJECTION') as reduced_e
**示例:**
```sql
SELECT id, VECTOR_NORMALIZE(embedding) as normalized_embedding FROM table
-```
\ No newline at end of file
+```
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf
index 3f0ad98b0b..192b6229e4 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf
@@ -54,7 +54,7 @@ transform {
sql {
plugin_input = "fake"
plugin_output = "fake1"
- query = "select ascii(c1) as c1_1, ascii(c2) as c2_1, bit_length(c4) as
c4_1, length(c4) as c4_2, octet_length(c4) as c4_3, char(c5) as c5_1,
concat(c1,id,'!') as c1_2, hextoraw(c6) as c6_1, rawtohex(c7) as c7_1,
insert(name,2,2,'**') as name1, lower(name) as name2, upper(name) as name3,
left(name, 3) as name4, right(name, 4) as name5, lpad(name, 10, '*') as name6,
rpad(name, 10, '*') as name7, ltrim(c8, '*') as c8_1, rtrim(c8, '*') as c8_2,
trim(c8, '*') as c8_3, regexp_replace(c9, [...]
+ query = "select ascii(c1) as c1_1, ascii(c2) as c2_1, bit_length(c4) as
c4_1, length(c4) as c4_2, octet_length(c4) as c4_3, char(c5) as c5_1,
concat(c1,id,'!') as c1_2, hextoraw(c6) as c6_1, rawtohex(c7) as c7_1,
insert(name,2,2,'**') as name1, lower(name) as name2, upper(name) as name3,
left(name, 3) as name4, right(name, 4) as name5, lpad(name, 10, '*') as name6,
rpad(name, 10, '*') as name7, ltrim(c8, '*') as c8_1, rtrim(c8, '*') as c8_2,
trim(c8, '*') as c8_3, regexp_replace(c9, [...]
}
}
@@ -295,8 +295,36 @@ sink {
rule_type = NOT_NULL
}
]
+ },
+ {
+ field_name = "name16"
+ field_type = "string"
+ field_value = [
+ {equals_to = "Sm95IERpbmc="}
+ ]
+ },
+ {
+ field_name = "name17"
+ field_type = "string"
+ field_value = [
+ {equals_to = "Joy Ding"}
+ ]
+ },
+ {
+ field_name = "name18"
+ field_type = "string"
+ field_value = [
+ {equals_to = "/v8ASgBvAHkAIABEAGkAbgBn"}
+ ]
+ },
+ {
+ field_name = "name19"
+ field_type = "string"
+ field_value = [
+ {equals_to = "Joy Ding"}
+ ]
}
]
}
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java
index e339434614..d6a21edc1e 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java
@@ -96,6 +96,8 @@ public class ZetaSQLFunction {
public static final String CONCAT_WS = "CONCAT_WS";
public static final String HEXTORAW = "HEXTORAW";
public static final String RAWTOHEX = "RAWTOHEX";
+ public static final String TO_BASE64 = "TO_BASE64";
+ public static final String FROM_BASE64 = "FROM_BASE64";
public static final String INSERT = "INSERT";
public static final String LOWER = "LOWER";
public static final String LCASE = "LCASE";
@@ -475,6 +477,10 @@ public class ZetaSQLFunction {
return StringFunction.hextoraw(args);
case RAWTOHEX:
return StringFunction.rawtohex(args);
+ case TO_BASE64:
+ return StringFunction.toBase64(args);
+ case FROM_BASE64:
+ return StringFunction.fromBase64(args);
case INSERT:
return StringFunction.insert(args);
case LOWER:
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
index 5988f0869d..7da359c8f5 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
@@ -327,6 +327,8 @@ public class ZetaSQLType {
case ZetaSQLFunction.CONCAT_WS:
case ZetaSQLFunction.HEXTORAW:
case ZetaSQLFunction.RAWTOHEX:
+ case ZetaSQLFunction.TO_BASE64:
+ case ZetaSQLFunction.FROM_BASE64:
case ZetaSQLFunction.INSERT:
case ZetaSQLFunction.LOWER:
case ZetaSQLFunction.LCASE:
diff --git
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunction.java
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunction.java
index 51fbe627db..ece5c3c7d1 100644
---
a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunction.java
+++
b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunction.java
@@ -22,12 +22,14 @@ import
org.apache.seatunnel.shade.com.google.common.hash.Hashing;
import org.apache.seatunnel.common.exception.CommonErrorCode;
import org.apache.seatunnel.common.utils.DateTimeUtils;
import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.EncodingUtils;
import org.apache.seatunnel.transform.exception.TransformException;
import org.apache.seatunnel.transform.sql.zeta.ZetaSQLFunction;
import org.apache.groovy.parser.antlr4.util.StringUtils;
import java.lang.reflect.Array;
+import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -36,6 +38,7 @@ import java.time.ZoneOffset;
import java.time.temporal.Temporal;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Base64;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@@ -189,6 +192,46 @@ public class StringFunction {
return buff.toString();
}
+ public static String toBase64(List<Object> args) {
+ if (args.size() > 2) {
+ throw new IllegalArgumentException("TO_BASE64 requires one or two
arguments");
+ }
+ Object arg = args.get(0);
+ if (arg == null) {
+ return null;
+ }
+ if (arg instanceof byte[]) {
+ if (args.size() == 2) {
+ throw new IllegalArgumentException(
+ "TO_BASE64 does not support charset for bytes input");
+ }
+ return Base64.getEncoder().encodeToString((byte[]) arg);
+ }
+ Charset charset = getBase64Charset(args);
+ return
Base64.getEncoder().encodeToString(arg.toString().getBytes(charset));
+ }
+
+ public static String fromBase64(List<Object> args) {
+ if (args.size() > 2) {
+ throw new IllegalArgumentException("FROM_BASE64 requires one or
two arguments");
+ }
+ Object arg = args.get(0);
+ if (arg == null) {
+ return null;
+ }
+ Charset charset = getBase64Charset(args);
+ try {
+ return new String(Base64.getDecoder().decode(arg.toString()),
charset);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Invalid Base64 content", e);
+ }
+ }
+
+ private static Charset getBase64Charset(List<Object> args) {
+ String charsetName = args.size() == 2 ? (String) args.get(1) : null;
+ return EncodingUtils.tryParseCharset(charsetName);
+ }
+
public static String insert(List<Object> args) {
String s1 = (String) args.get(0);
int start = ((Number) args.get(1)).intValue();
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLStringFunctionsTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLStringFunctionsTest.java
index e63102e19a..3c6c54a6d4 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLStringFunctionsTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLStringFunctionsTest.java
@@ -21,6 +21,7 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -29,6 +30,7 @@ import
org.apache.seatunnel.transform.exception.TransformException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -293,6 +295,79 @@ public class SQLStringFunctionsTest {
Assertions.assertEquals("010a", outRow.getField(0));
}
+ @Test
+ public void testToBase64() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"data"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+
+ SeaTunnelRow outRow =
+ runSql(
+ "select TO_BASE64(data) as encoded_data,"
+ + " TO_BASE64(data, 'UTF-16') as
utf16_encoded_data"
+ + " from dual",
+ rowType,
+ "SeaTunnel");
+
+ Assertions.assertEquals("U2VhVHVubmVs", outRow.getField(0));
+ Assertions.assertEquals("/v8AUwBlAGEAVAB1AG4AbgBlAGw=",
outRow.getField(1));
+ }
+
+ @Test
+ public void testFromBase64() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"data"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+
+ SeaTunnelRow outRow =
+ runSql(
+ "select FROM_BASE64(data) as decoded_data,"
+ + "
FROM_BASE64('/v8AUwBlAGEAVAB1AG4AbgBlAGw=', 'UTF-16') as utf16_decoded_data"
+ + " from dual",
+ rowType,
+ "U2VhVHVubmVs");
+
+ Assertions.assertEquals("SeaTunnel", outRow.getField(0));
+ Assertions.assertEquals("SeaTunnel", outRow.getField(1));
+ }
+
+ @Test
+ public void testToBase64WithBytesColumn() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"data"},
+ new SeaTunnelDataType[]
{PrimitiveByteArrayType.INSTANCE});
+
+ byte[] bytes = "SeaTunnel".getBytes(StandardCharsets.UTF_8);
+ SeaTunnelRow outRow =
+ runSql("select TO_BASE64(data) as encoded_data from dual",
rowType, bytes);
+
+ Assertions.assertEquals("U2VhVHVubmVs", outRow.getField(0));
+ }
+
+ @Test
+ public void testToBase64BytesColumnRejectsCharset() {
+ SeaTunnelRowType rowType =
+ new SeaTunnelRowType(
+ new String[] {"data"},
+ new SeaTunnelDataType[]
{PrimitiveByteArrayType.INSTANCE});
+
+ byte[] bytes = "SeaTunnel".getBytes(StandardCharsets.UTF_8);
+ TransformException exception =
+ Assertions.assertThrows(
+ TransformException.class,
+ () ->
+ runSql(
+ "select TO_BASE64(data, 'UTF-16') as
encoded_data from dual",
+ rowType,
+ bytes));
+
+ Assertions.assertInstanceOf(IllegalArgumentException.class,
exception.getCause());
+ Assertions.assertEquals(
+ "TO_BASE64 does not support charset for bytes input",
+ exception.getCause().getMessage());
+ }
+
@Test
public void testSoundex() {
SeaTunnelRowType rowType =
diff --git
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunctionTest.java
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunctionTest.java
index f0e52a2d87..00a858ce4e 100644
---
a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunctionTest.java
+++
b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/StringFunctionTest.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.transform.exception.TransformException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -29,6 +30,8 @@ import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.Temporal;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Date;
import java.util.List;
@@ -275,6 +278,114 @@ public class StringFunctionTest {
Assertions.assertEquals("010a", StringFunction.rawtohex(args));
}
+ @Test
+ public void testToBase64() {
+ List<Object> args = new ArrayList<>();
+ args.add("SeaTunnel");
+ Assertions.assertEquals("U2VhVHVubmVs", StringFunction.toBase64(args));
+
+ args.clear();
+ args.add("hello");
+ args.add("ISO-8859-1");
+ Assertions.assertEquals("aGVsbG8=", StringFunction.toBase64(args));
+
+ args.clear();
+ args.add("SeaTunnel");
+ args.add("UTF-16");
+ Assertions.assertEquals("/v8AUwBlAGEAVAB1AG4AbgBlAGw=",
StringFunction.toBase64(args));
+ }
+
+ @Test
+ public void testFromBase64() {
+ List<Object> args = new ArrayList<>();
+ args.add("U2VhVHVubmVs");
+ Assertions.assertEquals("SeaTunnel", StringFunction.fromBase64(args));
+
+ args.clear();
+ args.add("aGVsbG8=");
+ args.add("ISO-8859-1");
+ Assertions.assertEquals("hello", StringFunction.fromBase64(args));
+
+ args.clear();
+ args.add("/v8AUwBlAGEAVAB1AG4AbgBlAGw=");
+ args.add("UTF-16");
+ Assertions.assertEquals("SeaTunnel", StringFunction.fromBase64(args));
+ }
+
+ @Test
+ public void testToBase64WithBytesInput() {
+ List<Object> args = new ArrayList<>();
+ args.add("SeaTunnel".getBytes(StandardCharsets.UTF_8));
+ Assertions.assertEquals("U2VhVHVubmVs", StringFunction.toBase64(args));
+ }
+
+ @Test
+ public void testToBase64BytesInputRejectsCharset() {
+ IllegalArgumentException bytesWithCharset =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ StringFunction.toBase64(
+ Arrays.asList(
+
"SeaTunnel".getBytes(StandardCharsets.UTF_8),
+ "UTF-16")));
+ Assertions.assertEquals(
+ "TO_BASE64 does not support charset for bytes input",
+ bytesWithCharset.getMessage());
+ }
+
+ @Test
+ public void testBase64WithNullInputAndNullCharset() {
+ List<Object> args = new ArrayList<>();
+ args.clear();
+ args.add(null);
+ Assertions.assertNull(StringFunction.toBase64(args));
+ Assertions.assertNull(StringFunction.fromBase64(args));
+
+ Assertions.assertEquals(
+ "U2VhVHVubmVs",
StringFunction.toBase64(Arrays.asList("SeaTunnel", null)));
+ Assertions.assertEquals(
+ "SeaTunnel",
StringFunction.fromBase64(Arrays.asList("U2VhVHVubmVs", null)));
+ }
+
+ @Test
+ public void testBase64RejectsInvalidCharset() {
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> StringFunction.toBase64(Arrays.asList("SeaTunnel",
"invalid")));
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> StringFunction.fromBase64(Arrays.asList("U2VhVHVubmVs",
"invalid")));
+ }
+
+ @Test
+ public void testFromBase64RejectsInvalidContent() {
+ IllegalArgumentException invalidBase64 =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
StringFunction.fromBase64(Collections.singletonList("not-base64!")));
+ Assertions.assertEquals("Invalid Base64 content",
invalidBase64.getMessage());
+ }
+
+ @Test
+ public void testBase64RejectsInvalidArgumentCount() {
+ IllegalArgumentException invalidToBase64Args =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
StringFunction.toBase64(Arrays.asList("SeaTunnel", "UTF-8", "x")));
+ Assertions.assertEquals(
+ "TO_BASE64 requires one or two arguments",
invalidToBase64Args.getMessage());
+
+ IllegalArgumentException invalidFromBase64Args =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ StringFunction.fromBase64(
+ Arrays.asList("U2VhVHVubmVs", "UTF-8",
"x")));
+ Assertions.assertEquals(
+ "FROM_BASE64 requires one or two arguments",
invalidFromBase64Args.getMessage());
+ }
+
@Test
public void testInsertFunction() {
List<Object> args = new ArrayList<>();
