feat(transwarp): add Inceptor data source with table operations

This commit is contained in:
Hand Sonic
2026-09-29 02:05:07 +08:00
committed by GitHub
parent 597b134d21
commit 6fdc338348
101 changed files with 2300 additions and 66 deletions
+1
View File
@@ -1,6 +1,7 @@
**/libs/*.jar
!drivers/cache/libs/CacheDB.jar
!drivers/oscar/libs/oscarJDBC8.jar
!drivers/transwarp/libs/inceptor-sdk-transwarp-6.1.0-SNAPSHOT.jar
!drivers/uxdb/libs/uxdbjdbc-2.1.2.3p.jre8.jar
gradle.properties
build/
+1
View File
@@ -31,6 +31,7 @@ Each agent runs as a standalone process and communicates with DBX via stdin/stdo
| snowflake | Snowflake | Snowflake JDBC |
| trino | Trino (Presto) | Trino JDBC |
| hive | Apache Hive | Native Go HS2 agent |
| transwarp | Transwarp Inceptor | Bundled vendor JDBC agent |
| db2 | IBM DB2 | DB2 JDBC |
| informix | IBM Informix | Informix JDBC |
| neo4j | Neo4j | Official Neo4j Go Driver native agent |
+1
View File
@@ -31,6 +31,7 @@ DBX 的 Agent 驱动 —— 通过 JDBC 和原生数据库驱动支持各种数
| snowflake | Snowflake | Snowflake JDBC |
| trino | Trino (Presto) | Trino JDBC |
| hive | Apache Hive | 原生 Go HS2 agent |
| transwarp | 星环 Inceptor | 内置官方 JDBC 驱动 Agent |
| db2 | IBM DB2 | DB2 JDBC |
| informix | IBM Informix | Informix JDBC |
| neo4j | Neo4j | 官方 Neo4j Go Driver 原生 Agent |
+1 -1
View File
@@ -9,7 +9,7 @@ def pooledJdbcProjects = [
'firebird', 'gbase8a', 'gbase8s', 'goldendb', 'h2', 'h2-legacy', 'highgo',
'ignite', 'ignite3', 'informix', 'iris', 'cache', 'kylin', 'oceanbase-oracle', 'oscar', 'saphana',
'snowflake', 'spanner', 'spark', 'sqlserver-legacy', 'sundb', 'teradata', 'trino', 'uxdb',
'vertica', 'yashandb'
'vertica', 'yashandb', 'transwarp'
] as Set
def agentProjects = subprojects.findAll { !infrastructureProjects.contains(it.name) }
def jdbcAgentProjects = agentProjects.findAll { !legacyStandaloneProjects.contains(it.name) }
@@ -8,6 +8,7 @@ import com.google.gson.JsonParser;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -39,7 +40,7 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable {
private final ExecutorService cleanup;
private final JdbcConnectionPoolRegistry poolRegistry;
private final Gson gson = new Gson();
private final PrintStream protocolOutput = System.out;
private final PrintStream protocolOutput = new PrintStream(System.out, true, StandardCharsets.UTF_8);
private final Object outputLock = new Object();
private final Object maintenanceLock = new Object();
private final AtomicBoolean closed = new AtomicBoolean();
@@ -95,7 +96,7 @@ public final class MultiSessionJsonRpcServer implements AutoCloseable {
protocolOutput.println("{\"ready\":true}");
protocolOutput.flush();
}
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
JsonObject request = JsonParser.parseString(line).getAsJsonObject();
@@ -7,10 +7,11 @@ import com.google.gson.JsonParser;
import org.junit.jupiter.api.Test;
import java.sql.Connection;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.PrintStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
@@ -289,27 +290,55 @@ class CommonJavaCompatibilityTest {
}
@Test
void multiSessionServerKeepsProtocolOutputWhenGlobalStdoutChanges() {
void multiSessionServerUsesUtf8ProtocolWhenGlobalStdoutChanges() throws Exception {
synchronized (System.class) {
InputStream originalInput = System.in;
PrintStream originalOutput = System.out;
ByteArrayOutputStream protocolBytes = new ByteArrayOutputStream();
ByteArrayOutputStream redirectedBytes = new ByteArrayOutputStream();
try (PrintStream protocolOutput = new PrintStream(protocolBytes, true, StandardCharsets.UTF_8);
PrintStream redirectedOutput = new PrintStream(redirectedBytes, true, StandardCharsets.UTF_8)) {
try (PrintStream protocolOutput = new PrintStream(protocolBytes, true, java.nio.charset.Charset.forName("GBK"));
PrintStream redirectedOutput = new PrintStream(redirectedBytes, true, StandardCharsets.UTF_8);
PipedInputStream requestInput = new PipedInputStream();
PipedOutputStream requestWriter = new PipedOutputStream(requestInput)) {
System.setOut(protocolOutput);
MultiSessionJsonRpcServer server = new MultiSessionJsonRpcServer(MinimalAgent::new);
System.setIn(new ByteArrayInputStream(
"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"shutdown\",\"params\":{}}\n"
.getBytes(StandardCharsets.UTF_8)
));
MultiSessionJsonRpcServer server = MultiSessionJsonRpcServer.forSessionHandlers(() -> new SessionRpcHandler() {
@Override
public Object connect(JsonObject params) {
return Collections.singletonMap("label", "dbx\u4e2d\u6587");
}
@Override
public Object handle(String method, JsonObject params) {
return Collections.singletonMap("ok", true);
}
@Override
public void close() {
}
});
System.setIn(requestInput);
System.setOut(redirectedOutput);
server.run();
String protocol = protocolBytes.toString(StandardCharsets.UTF_8);
assertTrue(protocol.contains("{\"ready\":true}"), protocol);
assertTrue(protocol.contains("\"id\":1"), protocol);
Thread runner = new Thread(server::run, "dbx-utf8-protocol-test");
runner.setDaemon(true);
runner.start();
try {
requestWriter.write("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"test_connection\",\"params\":{}}\n"
.getBytes(StandardCharsets.UTF_8));
requestWriter.flush();
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
while (!protocolBytes.toString(StandardCharsets.UTF_8).contains("dbx\u4e2d\u6587")
&& runner.isAlive() && System.nanoTime() < deadline) {
Thread.sleep(10);
}
String protocol = protocolBytes.toString(StandardCharsets.UTF_8);
assertTrue(protocol.contains("{\"ready\":true}"), protocol);
assertTrue(protocol.contains("\"id\":1"), protocol);
assertTrue(protocol.contains("dbx\u4e2d\u6587"), protocol);
} finally {
requestWriter.close();
runner.join(TimeUnit.SECONDS.toMillis(5));
}
assertFalse(runner.isAlive(), "Protocol server did not stop after input closed");
assertEquals("", redirectedBytes.toString(StandardCharsets.UTF_8));
} finally {
System.setIn(originalInput);
+25
View File
@@ -0,0 +1,25 @@
dependencies {
implementation files('libs/inceptor-sdk-transwarp-6.1.0-SNAPSHOT.jar')
testImplementation 'com.h2database:h2:2.4.240'
}
tasks.named('shadowJar') {
doFirst {
if (!file('libs/inceptor-sdk-transwarp-6.1.0-SNAPSHOT.jar').isFile()) {
throw new GradleException('Missing bundled Transwarp SDK in libs/inceptor-sdk-transwarp-6.1.0-SNAPSHOT.jar')
}
}
manifest {
attributes(
'Agent-Label': 'Transwarp Inceptor',
'Main-Class': 'com.dbx.agent.transwarp.TranswarpAgent'
)
}
}
tasks.named('test') {
dependsOn tasks.named('shadowJar')
doFirst {
systemProperty 'dbx.transwarp.agent.jar', tasks.named('shadowJar').get().archiveFile.get().asFile.absolutePath
}
}
@@ -0,0 +1,443 @@
package com.dbx.agent.transwarp;
import com.dbx.agent.AbstractJdbcAgent;
import com.dbx.agent.ColumnInfo;
import com.dbx.agent.ConnectParams;
import com.dbx.agent.DatabaseInfo;
import com.dbx.agent.ForeignKeyInfo;
import com.dbx.agent.IndexInfo;
import com.dbx.agent.JdbcIdentifiers;
import com.dbx.agent.MetadataListConstraints;
import com.dbx.agent.MultiSessionJsonRpcServer;
import com.dbx.agent.ObjectInfo;
import com.dbx.agent.ObjectSource;
import com.dbx.agent.PartitionInfo;
import com.dbx.agent.QueryResult;
import com.dbx.agent.TableInfo;
import com.dbx.agent.TriggerInfo;
import org.apache.hive.jdbc.HiveConnection;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
public final class TranswarpAgent extends AbstractJdbcAgent {
private static final String DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver";
private boolean transactionStateUnknown;
@Override
protected void afterConnect(ConnectParams params, Connection connection) {
transactionStateUnknown = false;
}
@Override
protected void beforePooledConnectionReturn(Connection connection) throws Exception {
if (transactionStateUnknown) {
throw new SQLException("Transwarp transaction state is unknown after rollback failure");
}
}
@Override
protected String connectionValidationQuery() {
return "SELECT 1";
}
@Override
protected String driverClass() {
return DRIVER_CLASS;
}
@Override
protected String buildJdbcUrl(ConnectParams params) {
String supplied = params.getConnection_string();
if (supplied != null && !supplied.isBlank()) {
return supplied.trim();
}
String host = params.getHost() == null ? "" : params.getHost().trim();
if (host.isEmpty()) {
throw new IllegalArgumentException("Transwarp host is required");
}
int port = params.getPort() > 0 ? params.getPort() : 10000;
String database = database(params.getDatabase());
StringBuilder url = new StringBuilder("jdbc:inceptor2://")
.append(host).append(':').append(port).append('/').append(database);
String options = params.getUrl_params();
if (options != null && !options.isBlank()) {
url.append(';').append(options.replaceFirst("^;+", ""));
}
if (params.isSsl() && (options == null || !options.toLowerCase(Locale.ROOT).contains("ssl="))) {
url.append(";ssl=true");
}
return url.toString();
}
private static String database(String value) {
return value == null || value.isBlank() ? "default" : value.trim();
}
private String schema(String value) {
return value == null || value.isBlank() ? database(getConfiguredDatabase()) : value.trim();
}
@Override
public String setSchemaSQL(String schema) {
return schema == null || schema.isBlank() ? "" : "USE " + JdbcIdentifiers.INSTANCE.backtick(schema);
}
@Override
public List<DatabaseInfo> listDatabases() {
return query("SELECT database_name FROM system.databases_v ORDER BY database_name", List.of(),
row -> new DatabaseInfo(row.getString(1)));
}
@Override
public List<String> listSchemas() {
return listDatabases().stream().map(DatabaseInfo::getName).toList();
}
@Override
public List<TableInfo> listTables(String schema) {
return listTables(schema, MetadataListConstraints.NONE);
}
@Override
public List<TableInfo> listTables(String schema, MetadataListConstraints constraints) {
String catalog = schema(schema);
MetadataListConstraints requested = MetadataListConstraints.orNone(constraints);
List<TableInfo> tables = new ArrayList<>();
if (!requested.hasObjectTypes() || requested.tableTypeAllowed("TABLE") || requested.tableTypeAllowed("MATERIALIZED_VIEW")) {
tables.addAll(query(
"SELECT table_name, table_type, commentstring FROM system.tables_v WHERE database_name=? ORDER BY table_name",
List.of(catalog), row -> new TableInfo(row.getString(1), tableType(row.getString(2)), row.getString(3))));
}
if (!requested.hasObjectTypes() || requested.tableTypeAllowed("VIEW")) {
tables.addAll(query("SELECT view_name FROM system.views_v WHERE database_name=? ORDER BY view_name",
List.of(catalog), row -> new TableInfo(row.getString(1), "VIEW", null)));
}
tables.sort(Comparator.comparing(TableInfo::getName));
return requested.filterTables(tables);
}
private static String tableType(String value) {
if (value == null) return "TABLE";
String normalized = value.toUpperCase(Locale.ROOT).replace(' ', '_');
if (normalized.contains("MATERIALIZED") && normalized.contains("VIEW")) return "MATERIALIZED_VIEW";
return normalized.contains("VIEW") ? "VIEW" : "TABLE";
}
@Override
public List<ObjectInfo> listObjects(String schema) {
return listObjects(schema, MetadataListConstraints.NONE);
}
@Override
public List<ObjectInfo> listObjects(String schema, MetadataListConstraints constraints) {
String catalog = schema(schema);
MetadataListConstraints requested = MetadataListConstraints.orNone(constraints);
List<ObjectInfo> objects = new ArrayList<>();
if (requested.includesTableLikeTypes()) {
for (TableInfo table : listTables(catalog, requested.withoutPaging())) {
objects.add(new ObjectInfo(table.getName(), table.getTable_type(), catalog, table.getComment()));
}
}
if (!requested.hasObjectTypes() || requested.objectTypeAllowed("PROCEDURE")) {
objects.addAll(namedObjects(catalog, "system.procedures_v", "procedure_name", "PROCEDURE"));
}
if (!requested.hasObjectTypes() || requested.objectTypeAllowed("FUNCTION")) {
objects.addAll(namedObjects(catalog, "system.functions_v", "function_name", "FUNCTION"));
}
if (!requested.hasObjectTypes() || requested.objectTypeAllowed("PACKAGE") || requested.objectTypeAllowed("PACKAGE_BODY")) {
objects.addAll(optionalCatalogQuery(
"SELECT package_name, package_body FROM system.packages_v WHERE database_name=? ORDER BY package_name",
List.of(catalog), row -> {
List<ObjectInfo> entries = new ArrayList<>();
String name = row.getString(1);
entries.add(new ObjectInfo(name, "PACKAGE", catalog, null));
if (row.getString(2) != null) {
entries.add(new ObjectInfo(name, "PACKAGE_BODY", catalog, null));
}
return entries;
}).stream().flatMap(List::stream).toList());
}
objects.sort(Comparator.comparing(ObjectInfo::getName).thenComparing(ObjectInfo::getObject_type));
return requested.filterObjects(objects);
}
private List<ObjectInfo> namedObjects(String schema, String view, String nameColumn, String type) {
return optionalCatalogQuery("SELECT " + nameColumn + " FROM " + view + " WHERE database_name=? ORDER BY " + nameColumn,
List.of(schema), row -> new ObjectInfo(row.getString(1), type, schema, null));
}
@Override
public List<String> listDataTypes() {
return unchecked(() -> {
Set<String> types = new LinkedHashSet<>();
try (ResultSet rows = requireConnected().getMetaData().getTypeInfo()) {
while (rows.next()) {
String name = rows.getString("TYPE_NAME");
if (name != null && !name.isBlank()) types.add(name);
}
}
return types.stream().sorted().toList();
});
}
@Override
public List<ColumnInfo> getColumns(String schema, String table) {
String catalog = schema(schema);
Set<String> primaryKeys = unchecked(() -> {
Set<String> keys = new HashSet<>();
DatabaseMetaData metadata = requireConnected().getMetaData();
String metadataSchema = metadata.storesUpperCaseIdentifiers() ? catalog.toUpperCase(Locale.ROOT)
: metadata.storesLowerCaseIdentifiers() ? catalog.toLowerCase(Locale.ROOT) : catalog;
String metadataTable = metadata.storesUpperCaseIdentifiers() ? table.toUpperCase(Locale.ROOT)
: metadata.storesLowerCaseIdentifiers() ? table.toLowerCase(Locale.ROOT) : table;
try (ResultSet rows = metadata.getPrimaryKeys(null, metadataSchema, metadataTable)) {
while (rows.next()) keys.add(rows.getString("COLUMN_NAME").toLowerCase(Locale.ROOT));
} catch (SQLException ignored) {
// The system catalog still provides column metadata on SDKs without PK metadata.
}
return keys;
});
Map<String, int[]> dimensions = unchecked(() -> {
Map<String, int[]> sizes = new HashMap<>();
DatabaseMetaData metadata = requireConnected().getMetaData();
String metadataSchema = metadata.storesUpperCaseIdentifiers() ? catalog.toUpperCase(Locale.ROOT)
: metadata.storesLowerCaseIdentifiers() ? catalog.toLowerCase(Locale.ROOT) : catalog;
String metadataTable = metadata.storesUpperCaseIdentifiers() ? table.toUpperCase(Locale.ROOT)
: metadata.storesLowerCaseIdentifiers() ? table.toLowerCase(Locale.ROOT) : table;
try (ResultSet rows = metadata.getColumns(null, metadataSchema, metadataTable, null)) {
while (rows.next()) {
String column = rows.getString("COLUMN_NAME");
int size = rows.getInt("COLUMN_SIZE");
int scale = rows.getInt("DECIMAL_DIGITS");
sizes.put(column.toLowerCase(Locale.ROOT), new int[]{size, scale});
}
} catch (SQLException ignored) {
// The system catalog remains the source for type names and other attributes.
}
return sizes;
});
return query("SELECT column_name, column_type, nullable, default_value, commentstring "
+ "FROM system.columns_v WHERE database_name=? AND table_name=? ORDER BY column_id",
List.of(catalog, table), row -> {
String name = row.getString(1);
String type = row.getString(2);
int[] size = dimensions.get(name.toLowerCase(Locale.ROOT));
boolean numeric = type != null && type.matches("(?i)^(decimal|numeric|number)(\\(.*)?$");
boolean character = type != null && type.matches("(?i)^(char|varchar|character)(\\(.*)?$");
return new ColumnInfo(name, type, nullable(row.getObject(3)), row.getString(4),
primaryKeys.contains(name.toLowerCase(Locale.ROOT)), null, row.getString(5),
numeric && size != null ? size[0] : null, numeric && size != null ? size[1] : null,
character && size != null ? size[0] : null);
});
}
private static boolean nullable(Object value) {
if (value == null) return true;
if (value instanceof Boolean flag) return flag;
if (value instanceof Number number) return number.intValue() != 0;
String text = value.toString().trim().toUpperCase(Locale.ROOT);
return !text.equals("NO") && !text.equals("N") && !text.equals("FALSE") && !text.equals("0") && !text.equals("NOT NULL");
}
@Override
public List<IndexInfo> listIndexes(String schema, String table) {
// Waterdrop exposes no index list, and the SDK's getIndexInfo throws SQLException.
return List.of();
}
@Override
public List<ForeignKeyInfo> listForeignKeys(String schema, String table) {
return List.of();
}
@Override
public List<TriggerInfo> listTriggers(String schema, String table) {
String catalog = schema(schema);
return optionalCatalogQuery("SELECT * FROM system.triggers_v WHERE database_name=? AND table_name=?",
List.of(catalog, table), row -> new TriggerInfo(
firstColumn(row, "trigger_name", "name", "trigger"),
firstColumn(row, "event", "event_type", "trigger_event"),
firstColumn(row, "timing", "trigger_timing", "action_timing")))
.stream()
.filter(trigger -> trigger.getName() != null && !trigger.getName().isBlank())
.toList();
}
@Override
public List<PartitionInfo> listPartitions(String schema, String table) {
String catalog = schema(schema);
List<PartitionInfo> partitions = new ArrayList<>();
partitions.addAll(query("SELECT partition_name, partition_value, partition_key FROM system.partitions_v "
+ "WHERE database_name=? AND table_name=? ORDER BY partition_id",
List.of(catalog, table), row -> new PartitionInfo(row.getString(1), 0, row.getString(2), "LIST", row.getString(3))));
partitions.addAll(optionalCatalogQuery("SELECT partition_name, partition_range, partition_key FROM system.range_partitions_v "
+ "WHERE database_name=? AND table_name=? ORDER BY partition_id",
List.of(catalog, table), row -> new PartitionInfo(row.getString(1), 0, row.getString(2), "RANGE", row.getString(3))));
partitions.addAll(optionalCatalogQuery("SELECT * FROM system.buckets_v WHERE database_name=? AND table_name=?",
List.of(catalog, table), row -> new PartitionInfo(
firstColumn(row, "bucket_name", "name", "bucket"),
0,
firstColumn(row, "bucket_value", "value", "bucket_count", "count", "bucket_number"),
"BUCKET",
firstColumn(row, "bucket_key", "bucket_column", "column_name", "key"))));
List<PartitionInfo> numbered = new ArrayList<>();
for (int index = 0; index < partitions.size(); index++) {
PartitionInfo partition = partitions.get(index);
String name = partition.name();
if (name == null || name.isBlank()) {
name = partition.partition_type().equals("BUCKET") ? "bucket-" + (index + 1) : "partition-" + (index + 1);
}
numbered.add(new PartitionInfo(name, index + 1, partition.value(),
partition.partition_type(), partition.partition_key()));
}
return numbered;
}
@Override
public String getTableDdl(String schema, String table) {
String sql = "SHOW CREATE TABLE " + JdbcIdentifiers.INSTANCE.backtick(schema(schema))
+ "." + JdbcIdentifiers.INSTANCE.backtick(table);
return unchecked(() -> {
List<String> lines = new ArrayList<>();
try (Statement statement = requireConnected().createStatement(); ResultSet rows = statement.executeQuery(sql)) {
while (rows.next()) {
String line = rows.getString(1);
if (line != null && !line.isBlank()) lines.add(line);
}
}
return String.join("\n", lines);
});
}
@Override
public ObjectSource getObjectSource(String schema, String name, String objectType) {
String catalog = schema(schema);
String type = objectType.toUpperCase(Locale.ROOT);
if (type.equals("TABLE")) {
return new ObjectSource(name, type, catalog, getTableDdl(catalog, name));
}
String view;
String nameColumn;
String sourceColumn;
switch (type) {
case "VIEW" -> { view = "system.views_v"; nameColumn = "view_name"; sourceColumn = "origin_text"; }
case "PROCEDURE" -> { view = "system.procedures_v"; nameColumn = "procedure_name"; sourceColumn = "full_text"; }
case "FUNCTION" -> { view = "system.functions_v"; nameColumn = "function_name"; sourceColumn = "full_text"; }
case "PACKAGE" -> { view = "system.packages_v"; nameColumn = "package_name"; sourceColumn = "full_text"; }
case "PACKAGE_BODY" -> { view = "system.packages_v"; nameColumn = "package_name"; sourceColumn = "package_body"; }
default -> throw new UnsupportedOperationException("Unsupported Transwarp object type: " + type);
}
List<String> source = query("SELECT " + sourceColumn + " FROM " + view + " WHERE database_name=? AND "
+ nameColumn + "=?", List.of(catalog, name), row -> row.getString(1));
if (source.isEmpty() || source.stream().allMatch(value -> value == null || value.isBlank())) {
throw new IllegalArgumentException(type + " source not found: " + catalog + "." + name);
}
return new ObjectSource(name, type, catalog,
String.join("\n", source.stream().filter(value -> value != null && !value.isBlank()).toList()));
}
@Override
public QueryResult executeTransaction(List<String> statements, String schema) {
return unchecked(() -> {
long started = System.currentTimeMillis();
long affected = 0;
Connection connection = requireConnected();
applySchemaContext(connection, schema);
try (Statement statement = connection.createStatement()) {
if (connection instanceof HiveConnection) {
statement.execute("SET transaction.type = inceptor");
}
statement.execute("BEGIN TRANSACTION");
try {
for (String sql : statements) {
if (sql != null && !sql.isBlank()) {
statement.execute(sql);
affected += Math.max(statement.getUpdateCount(), 0);
}
}
statement.execute("COMMIT");
} catch (Exception error) {
try {
statement.execute("ROLLBACK");
} catch (Exception rollbackError) {
transactionStateUnknown = true;
error.addSuppressed(rollbackError);
}
throw error;
}
}
return new QueryResult(List.of(), List.of(), affected, System.currentTimeMillis() - started);
});
}
private interface RowReader<T> {
T read(ResultSet row) throws Exception;
}
private <T> List<T> query(String sql, List<String> parameters, RowReader<T> reader) {
return unchecked(() -> {
List<T> values = new ArrayList<>();
try (PreparedStatement statement = requireConnected().prepareStatement(sql)) {
for (int index = 0; index < parameters.size(); index++) {
statement.setString(index + 1, parameters.get(index));
}
try (ResultSet rows = statement.executeQuery()) {
while (rows.next()) values.add(reader.read(rows));
}
}
return values;
});
}
private <T> List<T> optionalCatalogQuery(String sql, List<String> parameters, RowReader<T> reader) {
try {
return query(sql, parameters, reader);
} catch (RuntimeException error) {
Throwable cause = error;
while (cause != null) {
if (cause instanceof SQLException sqlError) {
if ("42S02".equals(sqlError.getSQLState()) || "42P01".equals(sqlError.getSQLState())) {
return List.of();
}
break;
}
cause = cause.getCause();
}
throw error;
}
}
private static String firstColumn(ResultSet row, String... candidates) throws SQLException {
ResultSetMetaData metadata = row.getMetaData();
for (String candidate : candidates) {
for (int index = 1; index <= metadata.getColumnCount(); index++) {
String label = metadata.getColumnLabel(index);
String name = metadata.getColumnName(index);
if (candidate.equalsIgnoreCase(label) || candidate.equalsIgnoreCase(name)) {
Object value = row.getObject(index);
return value == null ? null : value.toString();
}
}
}
return null;
}
public static void main(String[] args) {
new MultiSessionJsonRpcServer(TranswarpAgent::new).run();
}
}
@@ -0,0 +1,234 @@
package com.dbx.agent.transwarp;
import com.dbx.agent.ColumnInfo;
import com.dbx.agent.ConnectParams;
import com.dbx.agent.MetadataListConstraints;
import com.dbx.agent.ObjectInfo;
import com.dbx.agent.PartitionInfo;
import com.dbx.agent.QueryPageOptions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.List;
import java.util.jar.JarFile;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class TranswarpAgentTest {
private static final String JDBC_URL = "jdbc:h2:mem:transwarp_agent;DB_CLOSE_DELAY=-1";
@BeforeAll
static void loadTestDatabaseDriver() throws Exception {
Class.forName("org.h2.Driver");
}
@Test
void buildsVendorUrlForBothProductProfiles() {
TranswarpAgent agent = new TranswarpAgent();
ConnectParams params = new ConnectParams("quark.example", 10000, "analytics", "user", "", "fetchSize=1000", "", false);
params.setSsl(true);
assertEquals("jdbc:inceptor2://quark.example:10000/analytics;fetchSize=1000;ssl=true", agent.buildJdbcUrl(params));
params.setConnection_string("jdbc:transwarp2://a:10000,b:10000/analytics;serviceDiscoveryMode=zooKeeper");
assertEquals(params.getConnection_string(), agent.buildJdbcUrl(params));
}
@Test
void unsupportedIndexAndForeignKeyMetadataDoesNotCallTheSdk() {
TranswarpAgent agent = new TranswarpAgent();
assertTrue(agent.listIndexes("demo", "orders").isEmpty());
assertTrue(agent.listForeignKeys("demo", "orders").isEmpty());
}
@Test
void readsWaterdropCatalogViewsWithoutHidingPackagesOrPartitions() throws Exception {
try (Connection connection = DriverManager.getConnection(JDBC_URL); Statement sql = connection.createStatement()) {
sql.execute("CREATE SCHEMA IF NOT EXISTS system");
sql.execute("CREATE SCHEMA IF NOT EXISTS demo");
sql.execute("DROP TABLE IF EXISTS demo.orders");
sql.execute("CREATE TABLE demo.orders (id INT PRIMARY KEY, amount DECIMAL(12,2))");
sql.execute("CREATE TABLE IF NOT EXISTS system.databases_v (database_name VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.tables_v (database_name VARCHAR, table_name VARCHAR, table_type VARCHAR, commentstring VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.views_v (database_name VARCHAR, view_name VARCHAR, origin_text VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.columns_v (database_name VARCHAR, table_name VARCHAR, column_id INT, column_name VARCHAR, column_type VARCHAR, nullable VARCHAR, default_value VARCHAR, commentstring VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.procedures_v (database_name VARCHAR, procedure_name VARCHAR, full_text VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.functions_v (database_name VARCHAR, function_name VARCHAR, full_text VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.packages_v (database_name VARCHAR, package_name VARCHAR, full_text VARCHAR, package_body VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.partitions_v (database_name VARCHAR, table_name VARCHAR, partition_id INT, partition_name VARCHAR, partition_value VARCHAR, partition_key VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.range_partitions_v (database_name VARCHAR, table_name VARCHAR, partition_id INT, partition_name VARCHAR, partition_range VARCHAR, partition_key VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.buckets_v (database_name VARCHAR, table_name VARCHAR, bucket_id INT, bucket_name VARCHAR, bucket_count INT, bucket_column VARCHAR)");
sql.execute("CREATE TABLE IF NOT EXISTS system.triggers_v (database_name VARCHAR, table_name VARCHAR, trigger_name VARCHAR, event VARCHAR, timing VARCHAR)");
sql.execute("INSERT INTO system.databases_v VALUES ('demo')");
sql.execute("INSERT INTO system.tables_v VALUES ('demo', 'orders', 'MANAGED_TABLE', 'sales')");
sql.execute("INSERT INTO system.tables_v VALUES ('demo', 'daily_mv', 'MATERIALIZED VIEW', NULL)");
sql.execute("INSERT INTO system.views_v VALUES ('demo', 'summary', 'CREATE VIEW summary AS SELECT id FROM orders')");
sql.execute("INSERT INTO system.columns_v VALUES ('demo', 'orders', 1, 'id', 'INT', 'false', NULL, 'order key')");
sql.execute("INSERT INTO system.columns_v VALUES ('demo', 'orders', 2, 'amount', 'DECIMAL(12,2)', 'true', '0', NULL)");
sql.execute("INSERT INTO system.procedures_v VALUES ('demo', 'load_orders', 'CREATE PROCEDURE load_orders() BEGIN END')");
sql.execute("INSERT INTO system.functions_v VALUES ('demo', 'total', 'CREATE FUNCTION total() RETURNS INT')");
sql.execute("INSERT INTO system.packages_v VALUES ('demo', 'reports', 'CREATE PACKAGE reports', 'CREATE PACKAGE BODY reports')");
sql.execute("INSERT INTO system.partitions_v VALUES ('demo', 'orders', 1, 'west', 'W', 'region')");
sql.execute("INSERT INTO system.range_partitions_v VALUES ('demo', 'orders', 2, 'recent', '2025', 'year')");
sql.execute("INSERT INTO system.buckets_v VALUES ('demo', 'orders', 1, 'bucket_0', 8, 'id')");
sql.execute("INSERT INTO system.triggers_v VALUES ('demo', 'orders', 'orders_audit', 'INSERT', 'AFTER')");
}
TranswarpAgent agent = connectedAgent();
try {
assertEquals(List.of("demo"), agent.listSchemas());
assertEquals("H2", agent.getDatabaseInfo().get("productName"));
assertEquals(List.of("daily_mv", "orders", "summary"), agent.listTables("demo").stream().map(table -> table.getName()).toList());
assertEquals(List.of("daily_mv"), agent.listTables("demo", new MetadataListConstraints(null, null, null, List.of("MATERIALIZED_VIEW")))
.stream().map(table -> table.getName()).toList());
List<ObjectInfo> objects = agent.listObjects("demo", MetadataListConstraints.NONE);
assertTrue(objects.stream().anyMatch(item -> item.getObject_type().equals("PACKAGE_BODY") && item.getName().equals("reports")));
assertTrue(objects.stream().anyMatch(item -> item.getObject_type().equals("PROCEDURE")));
assertEquals(List.of("reports"), agent.listObjects("demo", new MetadataListConstraints(null, null, null, List.of("PACKAGE")))
.stream().map(ObjectInfo::getName).toList());
List<ColumnInfo> columns = agent.getColumns("demo", "orders");
assertEquals(2, columns.size());
assertTrue(columns.get(0).getIs_primary_key());
assertFalse(columns.get(0).getIs_nullable());
assertEquals("0", columns.get(1).getColumn_default());
assertEquals(12, columns.get(1).getNumeric_precision());
assertEquals(2, columns.get(1).getNumeric_scale());
assertEquals("CREATE PACKAGE BODY reports", agent.getObjectSource("demo", "reports", "PACKAGE_BODY").getSource());
List<PartitionInfo> partitions = agent.listPartitions("demo", "orders");
assertEquals(3, partitions.size());
assertEquals("LIST", partitions.get(0).partition_type());
assertEquals("RANGE", partitions.get(1).partition_type());
assertEquals("BUCKET", partitions.get(2).partition_type());
assertEquals("bucket_0", partitions.get(2).name());
assertEquals("8", partitions.get(2).value());
assertEquals("id", partitions.get(2).partition_key());
assertEquals("orders_audit", agent.listTriggers("demo", "orders").get(0).getName());
try (Connection connection = DriverManager.getConnection(JDBC_URL); Statement sql = connection.createStatement()) {
sql.execute("DROP TABLE system.range_partitions_v");
sql.execute("DROP TABLE system.buckets_v");
sql.execute("DROP TABLE system.triggers_v");
sql.execute("DROP TABLE system.packages_v");
}
assertEquals(1, agent.listPartitions("demo", "orders").size());
assertTrue(agent.listObjects("demo").stream().anyMatch(item -> item.getName().equals("load_orders")));
} finally {
agent.disconnect();
}
}
@Test
void transactionFailureRollsBackEarlierStatement() throws Exception {
try (Connection connection = DriverManager.getConnection(JDBC_URL); Statement sql = connection.createStatement()) {
sql.execute("CREATE SCHEMA IF NOT EXISTS demo");
sql.execute("DROP TABLE IF EXISTS demo.tx_test");
sql.execute("CREATE TABLE demo.tx_test (id INT PRIMARY KEY)");
}
TranswarpAgent agent = connectedAgent();
try {
assertThrows(RuntimeException.class, () -> agent.executeTransaction(
List.of("INSERT INTO demo.tx_test VALUES (1)", "INSERT INTO demo.tx_test VALUES (1)"), ""));
try (Connection connection = DriverManager.getConnection(JDBC_URL);
var sql = connection.createStatement();
var rows = sql.executeQuery("SELECT COUNT(*) FROM demo.tx_test")) {
assertTrue(rows.next());
assertEquals(0, rows.getInt(1));
}
} finally {
agent.disconnect();
}
}
@Test
void pagesQueryResultsThroughSharedJdbcExecutor() throws Exception {
try (Connection connection = DriverManager.getConnection(JDBC_URL); Statement sql = connection.createStatement()) {
sql.execute("CREATE SCHEMA IF NOT EXISTS demo");
sql.execute("DROP TABLE IF EXISTS demo.page_test");
sql.execute("CREATE TABLE demo.page_test (id INT PRIMARY KEY)");
sql.execute("INSERT INTO demo.page_test VALUES (1), (2), (3)");
}
TranswarpAgent agent = connectedAgent();
try {
var first = agent.executeQueryPage("SELECT id FROM demo.page_test ORDER BY id", "",
new QueryPageOptions(2, null, 10));
assertEquals(2, first.getRows().size());
assertTrue(first.getHas_more());
var next = agent.fetchQueryPage(first.getSession_id(), 2);
assertEquals(1, next.getRows().size());
assertFalse(next.getHas_more());
} finally {
agent.disconnect();
}
}
@Test
void packagedAgentLoadsDefaultDriverWithoutExternalClasspath() throws Exception {
Path jar = Path.of(System.getProperty("dbx.transwarp.agent.jar"));
assertTrue(Files.isRegularFile(jar));
try (JarFile archive = new JarFile(jar.toFile())) {
assertFalse(Boolean.parseBoolean(archive.getManifest().getMainAttributes().getValue("Agent-External-Driver")));
}
try (URLClassLoader loader = new URLClassLoader(new URL[]{jar.toUri().toURL()}, ClassLoader.getPlatformClassLoader())) {
Class<?> paramsClass = loader.loadClass("com.dbx.agent.ConnectParams");
Object params = paramsClass.getDeclaredConstructor().newInstance();
Object agent = loader.loadClass("com.dbx.agent.transwarp.TranswarpAgent").getDeclaredConstructor().newInstance();
var loadDriver = loader.loadClass("com.dbx.agent.AbstractJdbcAgent").getDeclaredMethod("loadDriver", paramsClass);
loadDriver.setAccessible(true);
loadDriver.invoke(agent, params);
Driver driver = (Driver) Class.forName("org.apache.hive.jdbc.HiveDriver", true, loader)
.getDeclaredConstructor().newInstance();
assertTrue(driver.acceptsURL("jdbc:inceptor2://localhost:10000/default"));
assertTrue(driver.acceptsURL("jdbc:transwarp2://localhost:10000/default"));
assertTrue(driver.acceptsURL("jdbc:hive2://localhost:10000/default"));
}
}
@Test
void validatesUsingSharedJdbcConnectionLifecycle() {
ConnectParams params = new ConnectParams("", 0, "demo", "", "", "", JDBC_URL, false);
params.setJdbc_driver_class("org.h2.Driver");
var result = new TranswarpAgent().testConnectionWithInfo(params);
assertEquals(true, result.get("ok"));
assertEquals("SELECT 1", result.get("validation"));
}
@Test
void packagedAgentRunsSharedJsonRpcProtocol() throws Exception {
String java = Path.of(System.getProperty("java.home"), "bin", "java").toString();
Process process = new ProcessBuilder(java, "-jar", System.getProperty("dbx.transwarp.agent.jar"))
.redirectErrorStream(true).start();
try {
try (var input = process.getOutputStream()) {
input.write("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"shutdown\",\"params\":{}}\n".getBytes(StandardCharsets.UTF_8));
}
assertTrue(process.waitFor(10, TimeUnit.SECONDS), "Packaged Agent did not shut down");
String output = new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
assertEquals(0, process.exitValue(), output);
assertTrue(output.contains("\"ready\":true"), output);
assertTrue(output.contains("\"result\":"), output);
assertFalse(output.contains("\"error\":"), output);
} finally {
process.destroyForcibly();
process.waitFor(5, TimeUnit.SECONDS);
}
}
private static TranswarpAgent connectedAgent() {
ConnectParams params = new ConnectParams("", 0, "demo", "", "", "", JDBC_URL, false);
params.setJdbc_driver_class("org.h2.Driver");
params.setDriver_profile("transwarp-inceptor");
TranswarpAgent agent = new TranswarpAgent();
agent.connect(params);
return agent;
}
}
+1
View File
@@ -29,4 +29,5 @@ spark intentional-fallback java-show-metadata Uses SHOW TABLES/SHOW DATABASES; s
sundb native-pushdown java-sql Uses information_schema with type, filter, stable order, and LIMIT/OFFSET.
tdengine shared-fallback native-rust Uses SHOW STABLES/TABLES with information_schema parent enrichment, then applies stable filtering and paging in the native agent.
trino native-pushdown java-sql Uses information_schema.tables with type, filter, stable order, and literal LIMIT/OFFSET.
transwarp shared-fallback java-sql Uses system catalog views with stable ordering and shared filtering/paging; the vendor SDK has no verified portable server-side paging contract.
xugu native-pushdown native-go Uses ALL_TABLES/ALL_VIEWS metadata with type, filter, stable order, and ROWNUM paging.
1 driver strategy scope reason
29 sundb native-pushdown java-sql Uses information_schema with type, filter, stable order, and LIMIT/OFFSET.
30 tdengine shared-fallback native-rust Uses SHOW STABLES/TABLES with information_schema parent enrichment, then applies stable filtering and paging in the native agent.
31 trino native-pushdown java-sql Uses information_schema.tables with type, filter, stable order, and literal LIMIT/OFFSET.
32 transwarp shared-fallback java-sql Uses system catalog views with stable ordering and shared filtering/paging; the vendor SDK has no verified portable server-side paging contract.
33 xugu native-pushdown native-go Uses ALL_TABLES/ALL_VIEWS metadata with type, filter, stable order, and ROWNUM paging.
+1 -1
View File
@@ -6,7 +6,7 @@ def driverModules = [
'teradata', 'vertica', 'firebird', 'exasol', 'oceanbase-oracle', 'gbase8a', 'gbase8s',
'bigquery', 'kylin', 'ignite', 'ignite3', 'spanner', 'sundb', 'h2', 'h2-legacy', 'snowflake', 'trino', 'spark',
'db2', 'informix', 'mongodb', 'highgo', 'uxdb', 'yashandb', 'oscar',
'iris', 'cache', 'kafka', 'sqlserver-legacy'
'iris', 'cache', 'kafka', 'sqlserver-legacy', 'transwarp'
]
include(*(infrastructureModules + driverModules))
+1
View File
@@ -34,6 +34,7 @@
"teradata": "0.1.63",
"tdengine": "0.1.50",
"trino": "0.1.68",
"transwarp": "0.1.0",
"vastbase": "0.1.51",
"vertica": "0.1.63",
"mongodb": "0.1.71",
Binary file not shown.

After

Width:  |  Height:  |  Size: 2.2 KiB

@@ -2885,6 +2885,9 @@ function applyProfile(val: string, preserveConnectionFields = false) {
jdbcDriverPathsInput.value = "";
jdbcManualClasspathOpen.value = true;
}
if (profile.type === "transwarp") {
form.value.connection_string = undefined;
}
if (profile.type === "spanner") {
// Google Cloud endpoints carry no host; the local emulator is opted into
// by typing host `localhost` and port 9010 explicitly.
@@ -2940,7 +2943,7 @@ function applyProfile(val: string, preserveConnectionFields = false) {
if (profile.type === "salesforce") {
resetSalesforceOAuthFields(form.value.external_config, form.value.password);
}
resetHiveKerberosFields(profile.type === "hive" || profile.type === "argo" || profile.type === "kyuubi" || profile.type === "impala" ? form.value : undefined);
resetHiveKerberosFields(profile.type === "hive" || profile.type === "argo" || profile.type === "transwarp" || profile.type === "kyuubi" || profile.type === "impala" ? form.value : undefined);
}
if (profile.type === "meilisearch") {
syncMeilisearchHostInput(form.value);
@@ -3174,7 +3177,7 @@ watch(
resetSalesforceOAuthFields(undefined, undefined);
}
resetElasticsearchProxyFields(config.db_type === "elasticsearch" ? config.external_config : undefined);
resetHiveKerberosFields(config.db_type === "hive" || config.db_type === "argo" || config.db_type === "kyuubi" || config.db_type === "impala" ? config : undefined);
resetHiveKerberosFields(config.db_type === "hive" || config.db_type === "argo" || config.db_type === "transwarp" || config.db_type === "kyuubi" || config.db_type === "impala" ? config : undefined);
resetDamengJvmOptions(config.db_type === "dameng" ? config : undefined);
h2ConnectionMode.value = h2ConnectionModeForConfig(config);
customColorInput.value = config.color || "";
@@ -4696,7 +4699,7 @@ function connectionConfigForSubmit(id: string, generatedName = "", validatePlugi
config.ssl = !!config.ssl || damengSsl.enabled;
config.url_params = applyDamengSslUrlParams(config.url_params, config.ssl, damengSsl.sslFilesPath, damengSsl.sslKeystorePassword, damengSsl.sslProtocol);
}
if (config.db_type === "hive" || config.db_type === "argo" || config.db_type === "kyuubi" || config.db_type === "impala") {
if (config.db_type === "hive" || config.db_type === "argo" || config.db_type === "transwarp" || config.db_type === "kyuubi" || config.db_type === "impala") {
if (hiveAuthMode.value === "kerberos" && !hivePrincipal.value.trim()) {
throw new Error(t("connection.hiveKerberosPrincipalRequired"));
}
@@ -5093,6 +5096,10 @@ function connectionConfigForSubmit(id: string, generatedName = "", validatePlugi
config.jdbc_driver_class = undefined;
config.jdbc_driver_paths = [];
}
if (config.db_type === "transwarp") {
config.jdbc_driver_class = undefined;
config.jdbc_driver_paths = [];
}
if (config.db_type === "h2") {
if (config.driver_profile === "h2-custom") {
config.jdbc_driver_class = config.jdbc_driver_class?.trim() || "org.h2.Driver";
@@ -8880,7 +8887,7 @@ function openExternalUrl(url: string) {
<p class="col-span-3 text-xs text-muted-foreground">{{ t("connection.oracleTnsPathHint") }}</p>
</div>
<template v-if="form.db_type === 'hive' || form.db_type === 'kyuubi' || form.db_type === 'impala'">
<template v-if="form.db_type === 'hive' || form.db_type === 'transwarp' || form.db_type === 'kyuubi' || form.db_type === 'impala'">
<div class="grid grid-cols-4 items-center gap-4">
<Label :class="connectionLabelClass">{{ t("connection.hiveAuthMode") }}</Label>
<div class="col-span-3 grid h-8 grid-cols-2 overflow-hidden rounded-md border border-input bg-muted/30 p-0.5">
@@ -9023,7 +9030,9 @@ function openExternalUrl(url: string) {
? 'catalog=paimon_catalog'
: form.db_type === 'cassandra'
? 'localdatacenter=dc1'
: 'sslmode=prefer'
: form.db_type === 'transwarp'
? 'fetchSize=500;auth=noSasl'
: 'sslmode=prefer'
"
/>
<p v-if="showGenericUrlParamsHint" class="text-xs leading-5 text-muted-foreground">
@@ -3409,13 +3409,14 @@ const canDeleteExistingRows = computed(() => !!props.customSaveHandler || canDel
watch(
() => [props.databaseType, props.connectionId, props.database, props.tableMeta?.schema, props.tableMeta?.tableName],
async () => {
if (props.databaseType !== "hive" || !props.connectionId || !props.database || !props.tableMeta) {
if ((props.databaseType !== "hive" && props.databaseType !== "transwarp") || !props.connectionId || !props.database || !props.tableMeta) {
hiveTableTransactional.value = undefined;
return;
}
try {
const sql = await buildHiveTablePropertiesSql({
schema: props.tableMeta.schema,
databaseType: props.databaseType,
schema: props.tableMeta.schema || props.database,
tableName: props.tableMeta.tableName,
propertyName: "transactional",
});
@@ -66,6 +66,11 @@ describe("ConnectionIcon", () => {
expect(listPlugins).not.toHaveBeenCalled();
});
it("shows the Inceptor logo for a saved Transwarp connection", async () => {
const container = await mountIcon({ ...connection, db_type: "transwarp", driver_profile: "transwarp-inceptor" });
expect(container.querySelector("img")?.getAttribute("src")).toBe("/icons/database/transwarp-inceptor.png");
});
it("renders a safe fallback for missing connection metadata", async () => {
const container = await mountIcon();
expect(container.querySelector("img")?.getAttribute("src")).toBe("/icons/database/postgres.svg");
+11
View File
@@ -55,6 +55,17 @@ describe("DatabaseIcon", () => {
app.unmount();
});
it.each(["transwarp", "transwarp-inceptor"])("uses the Inceptor asset for %s", async (dbType) => {
const container = document.createElement("div");
document.body.appendChild(container);
const app = createApp(DatabaseIcon, { dbType });
app.mount(container);
await nextTick();
expect(container.querySelector("img")?.getAttribute("src")).toBe("/icons/database/transwarp-inceptor.png");
app.unmount();
});
it("uses the Meilisearch asset", async () => {
const container = document.createElement("div");
document.body.appendChild(container);
@@ -60,6 +60,8 @@ const assetIcons: Record<string, string> = {
prestosql: "presto",
hive: "hive",
argo: "hive",
transwarp: "transwarp-inceptor.png",
transwarp_inceptor: "transwarp-inceptor.png",
kyuubi: "kyuubi.png",
impala: "impala",
hbase: "hbase",
@@ -129,6 +131,7 @@ const assetIcons: Record<string, string> = {
const normalizedType = computed(() => (props.dbType || "").toLowerCase().replace(/[\s-]+/g, "_"));
const assetName = computed(() => assetIcons[normalizedType.value]);
const useLightIconInDarkMode = computed(() => normalizedType.value === "easysearch" && isDark.value);
const brightenInceptorInDarkMode = computed(() => isDark.value && (normalizedType.value === "transwarp" || normalizedType.value === "transwarp_inceptor"));
const assetSrc = computed(() => {
if (!assetName.value) return "";
if (normalizedType.value === "uxdb" && isDark.value) return webPath("/icons/database/uxdb-dark.svg");
@@ -137,7 +140,14 @@ const assetSrc = computed(() => {
</script>
<template>
<img v-if="assetName" :src="assetSrc" alt="" class="database-logo object-contain" :class="{ 'database-logo-light': useLightIconInDarkMode, 'database-logo-impala': normalizedType === 'impala', 'database-logo-solr': normalizedType === 'solr' }" aria-hidden="true" />
<img
v-if="assetName"
:src="assetSrc"
alt=""
class="database-logo object-contain"
:class="{ 'database-logo-light': useLightIconInDarkMode, 'database-logo-inceptor-dark': brightenInceptorInDarkMode, 'database-logo-impala': normalizedType === 'impala', 'database-logo-solr': normalizedType === 'solr' }"
aria-hidden="true"
/>
<Database v-else class="text-blue-400" />
</template>
@@ -151,6 +161,10 @@ const assetSrc = computed(() => {
filter: brightness(0) invert(82%);
}
.database-logo-inceptor-dark {
filter: brightness(1.6);
}
.database-logo-impala {
transform: scale(1.55);
}
@@ -0,0 +1,98 @@
<script setup lang="ts">
import { computed, ref } from "vue";
import { ChevronDown } from "@lucide/vue";
import { useI18n } from "vue-i18n";
import { Button } from "@/components/ui/button";
import { Input } from "@/components/ui/input";
import { DropdownMenu, DropdownMenuCheckboxItem, DropdownMenuContent, DropdownMenuTrigger } from "@/components/ui/dropdown-menu";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
import type { TablePhysicalOptionsConfig, TablePhysicalOptionsDraft } from "@/lib/table/tablePhysicalOptions";
const props = defineProps<{
modelValue: TablePhysicalOptionsDraft;
config: TablePhysicalOptionsConfig;
columns: Array<{ id: string; name: string }>;
disabled?: boolean;
}>();
const emit = defineEmits<{ "update:modelValue": [value: TablePhysicalOptionsDraft] }>();
const { t } = useI18n();
const search = ref({ partition: "", distribution: "" });
type ColumnGroup = "partition" | "distribution";
const columnGroups = computed(() => [
{ key: "partition" as const, label: props.config.partitionColumnsLabel },
{ key: "distribution" as const, label: props.config.distributionColumnsLabel },
]);
function selectedIds(group: ColumnGroup): string[] {
return group === "partition" ? props.modelValue.partitionColumnIds : props.modelValue.distributionColumnIds;
}
function selectedNames(group: ColumnGroup): string {
const selected = new Set(selectedIds(group));
return props.columns
.filter((column) => selected.has(column.id))
.map((column) => column.name)
.join(", ");
}
function availableColumns(group: ColumnGroup) {
const filter = search.value[group].trim().toLowerCase();
return props.columns.filter((column) => !filter || column.name.toLowerCase().includes(filter));
}
function toggleColumn(group: ColumnGroup, id: string) {
const current = selectedIds(group);
const next = current.includes(id) ? current.filter((selected) => selected !== id) : [...current, id];
emit("update:modelValue", {
...props.modelValue,
[group === "partition" ? "partitionColumnIds" : "distributionColumnIds"]: next,
});
}
function updateValue<K extends keyof TablePhysicalOptionsDraft>(field: K, value: TablePhysicalOptionsDraft[K]) {
emit("update:modelValue", { ...props.modelValue, [field]: value });
}
</script>
<template>
<div class="flex shrink-0 flex-wrap items-end gap-3 border-b pb-2">
<label v-for="group in columnGroups" :key="group.key" class="flex min-w-0 max-w-full flex-col gap-1 text-muted-foreground">
<span>{{ t(group.label) }}</span>
<DropdownMenu>
<DropdownMenuTrigger as-child>
<Button variant="outline" size="sm" class="h-8 w-48 max-w-full justify-between gap-1 px-2 font-normal" :disabled="disabled" :aria-label="t(group.label)">
<span class="min-w-0 truncate font-mono">{{ selectedNames(group.key) || t("structureEditor.indexColumnsPlaceholder") }}</span>
<ChevronDown class="size-3.5 shrink-0" />
</Button>
</DropdownMenuTrigger>
<DropdownMenuContent class="max-h-64 min-w-48 overflow-y-auto" side="bottom" :side-offset="2">
<div class="px-2 pb-1 pt-0.5">
<Input v-model="search[group.key]" class="h-7 text-xs" :placeholder="t('grid.search')" @click.stop />
</div>
<DropdownMenuCheckboxItem v-for="column in availableColumns(group.key)" :key="column.id" :checked="selectedIds(group.key).includes(column.id)" @select.prevent @click="toggleColumn(group.key, column.id)">
{{ column.name }}
</DropdownMenuCheckboxItem>
<div v-if="availableColumns(group.key).length === 0" class="px-2 py-1 text-xs text-muted-foreground">{{ t("structureEditor.indexColumnsPlaceholder") }}</div>
</DropdownMenuContent>
</DropdownMenu>
</label>
<label class="flex min-w-0 max-w-full flex-col gap-1 text-muted-foreground">
<span>{{ t(config.bucketCountLabel) }}</span>
<Input :model-value="modelValue.bucketCount" type="number" min="1" :max="config.maxBuckets" step="1" class="h-8 w-24 max-w-full" :disabled="disabled" @update:model-value="updateValue('bucketCount', String($event ?? ''))" />
</label>
<label v-if="config.storageFormats.length" class="flex min-w-0 max-w-full flex-col gap-1 text-muted-foreground">
<span>{{ t(config.storageFormatLabel) }}</span>
<Select :model-value="modelValue.storageFormat || '__default'" :disabled="disabled" @update:model-value="updateValue('storageFormat', $event === '__default' ? '' : String($event))">
<SelectTrigger class="w-32 max-w-full"><SelectValue :placeholder="t('structureEditor.defaultAction')" /></SelectTrigger>
<SelectContent>
<SelectItem value="__default">{{ t("structureEditor.defaultAction") }}</SelectItem>
<SelectItem v-for="format in config.storageFormats" :key="format" :value="format">{{ format }}</SelectItem>
</SelectContent>
</Select>
</label>
<label v-if="config.transactionalLabel" class="flex h-9 items-center gap-2 text-muted-foreground">
<input :checked="modelValue.transactional" type="checkbox" class="h-4 w-4 accent-primary" :disabled="disabled" @change="updateValue('transactional', ($event.target as HTMLInputElement).checked)" />
{{ t(config.transactionalLabel) }}
</label>
</div>
</template>
@@ -16,6 +16,7 @@ import { DropdownMenu, DropdownMenuCheckboxItem, DropdownMenuContent, DropdownMe
import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip";
import { SearchableSelect } from "@/components/ui/searchable-select";
import TablePhysicalOptionsEditor from "@/components/structure/TablePhysicalOptionsEditor.vue";
import CustomContextMenu, { type ContextMenuItem } from "@/components/ui/CustomContextMenu.vue";
import EditorSearchPanel from "@/components/editor/EditorSearchPanel.vue";
import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
@@ -58,6 +59,7 @@ import {
import { buildMysqlAutoIncrementCounterStatement, canEditMysqlAutoIncrementCounter, refreshMysqlAutoIncrementCounterDraft } from "@/lib/table/mysqlAutoIncrementCounter";
import { mysqlTableCollationSql, parseMysqlTableCollation } from "@/lib/table/mysqlTableCollation";
import { MYSQL_STORAGE_ENGINES_SQL, mysqlTableEngineSql, mysqlTableEngineSqlOption, parseMysqlTableEngineMetadata, refreshMysqlTableEngineDraft, supportsMysqlTableEngine } from "@/lib/table/mysqlTableEngine";
import { INCEPTOR_PHYSICAL_OPTIONS, buildInceptorCreateOptions, emptyTablePhysicalOptions, hasTablePhysicalOptions, pruneTablePhysicalOptions, restoreTablePhysicalOptions } from "@/lib/table/tablePhysicalOptions";
import { PRESET_FIELDS_TEMPLATE_ID, createTableColumnTemplateDrafts } from "@/lib/table/tableColumnTemplates";
import { getMysqlDataTypeHelp } from "@/lib/table/mysqlDataTypeHelp";
import { getPostgresDataTypeHelp, gaussdbMTypeDisplayName } from "@/lib/table/postgresDataTypeHelp";
@@ -1581,6 +1583,14 @@ const mysqlAutoIncrementLoading = ref(false);
const mysqlAutoIncrementLoadError = ref("");
const mysqlTableEngine = ref("");
const originalMysqlTableEngine = ref("");
const physicalOptions = ref(emptyTablePhysicalOptions());
watch(
() => columns.value.filter((column) => !column.markedForDrop).map((column) => column.id),
(ids) => {
if (!isCreateMode.value) return;
physicalOptions.value = pruneTablePhysicalOptions(physicalOptions.value, new Set(ids));
},
);
const mysqlTableEngineOptions = ref<string[]>([]);
const mysqlTableEngineLoading = ref(false);
const mysqlTableEngineLoadError = ref("");
@@ -1900,6 +1910,7 @@ function createCurrentDraft(initialized = true): TableStructureEditorDraft {
originalMysqlAutoIncrementValue: originalMysqlAutoIncrementValue.value,
mysqlTableEngine: mysqlTableEngine.value,
originalMysqlTableEngine: originalMysqlTableEngine.value,
physicalOptions: cloneDraftValue(physicalOptions.value),
tableOwner: tableOwner.value,
originalTableOwner: originalTableOwner.value,
columns: cloneDraftValue(columns.value),
@@ -1977,6 +1988,7 @@ function restoreDraft(draft: TableStructureEditorDraft) {
tableOwner.value = draft.tableOwner || "";
originalTableOwner.value = draft.originalTableOwner || "";
columns.value = cloneDraftValue(draft.columns || []);
physicalOptions.value = restoreTablePhysicalOptions(draft);
// Existing-index edits never support Concurrent (the checkbox is disabled and
// the core builder rejects the request), so a stale `concurrently: true`
// saved in a restored draft must not be submitted or deadlock the save.
@@ -2066,6 +2078,7 @@ function hasPendingStructureChanges(): boolean {
!!newTableName.value.trim() ||
!!tableComment.value.trim() ||
mysqlTableEngine.value !== originalMysqlTableEngine.value ||
hasTablePhysicalOptions(physicalOptions.value) ||
columns.value.length > 0 ||
indexes.value.length > 0 ||
foreignKeys.value.length > 0 ||
@@ -2236,6 +2249,7 @@ function structureChangeOptions(): BuildTableStructureChangeSqlOptions {
tableComment: tableComment.value,
originalTableComment: isCreateMode.value ? undefined : originalTableComment.value,
mysqlEngine: mysqlTableEngineSqlOption({ value: mysqlTableEngine.value, originalValue: originalMysqlTableEngine.value }, isCreateMode.value, supportsMysqlEngine.value && !mysqlTableEngineLoading.value && !mysqlTableEngineLoadError.value),
transwarpCreate: isCreateMode.value && databaseType.value === "transwarp" ? buildInceptorCreateOptions(physicalOptions.value, columns.value) : undefined,
tableCollation: mysqlTableDefaultCollation.value || undefined,
partitioned: isPartitionedParent.value,
isGaussdbMMode: connection.value?.driver_profile?.toLowerCase() === "gaussdb-m",
@@ -2425,6 +2439,7 @@ function resetState() {
mysqlAutoIncrementLoadError.value = "";
mysqlTableEngine.value = "";
originalMysqlTableEngine.value = "";
physicalOptions.value = emptyTablePhysicalOptions();
mysqlTableEngineOptions.value = [];
mysqlTableEngineLoadRequestId += 1;
mysqlTableEngineLoading.value = false;
@@ -4786,6 +4801,7 @@ watch(
mysqlAutoIncrementLoadError,
mysqlTableEngine,
originalMysqlTableEngine,
physicalOptions,
mysqlTableEngineLoading,
mysqlTableEngineLoadError,
mysqlTableDefaultCollation,
@@ -4965,6 +4981,8 @@ watch(
</Tooltip>
</div>
<TablePhysicalOptionsEditor v-if="isCreateMode && databaseType === 'transwarp'" v-model="physicalOptions" :config="INCEPTOR_PHYSICAL_OPTIONS" :columns="columns.filter((column) => !column.markedForDrop && !!column.name.trim())" :disabled="saving" />
<div v-if="supportsTableOwner" class="flex shrink-0 items-center gap-2">
<label class="flex shrink-0 items-center gap-1 font-medium text-muted-foreground">
<UserRound :class="structureIconClass" />
+5
View File
@@ -5277,6 +5277,11 @@ export default {
mysqlTableEngineSearchPlaceholder: "Search storage engines",
mysqlTableEngineEmpty: "No storage engines available",
mysqlTableEngineLoadFailed: "Failed to load storage engines: {message}",
transwarpPartitionColumns: "Partition columns",
transwarpBucketColumns: "Bucket columns",
transwarpBucketCount: "Buckets",
transwarpStorageFormat: "Storage format",
transwarpTransactional: "Transactional table",
owner: "Owner",
ownerPlaceholder: "Enter role name...",
ownerSearchPlaceholder: "Search or enter a role name...",
+5
View File
@@ -4773,6 +4773,11 @@ export default withEnglishFallback({
onDelete: "Al eliminar",
onUpdate: "Al actualizar",
defaultAction: "Predeterminado",
transwarpPartitionColumns: "Columnas de partición",
transwarpBucketColumns: "Columnas de distribución",
transwarpBucketCount: "Número de buckets",
transwarpStorageFormat: "Formato de almacenamiento",
transwarpTransactional: "Tabla transaccional",
triggerName: "Disparador",
triggerStatement: "Sentencia del disparador",
primary: "Primaria",
+5
View File
@@ -4695,6 +4695,11 @@ export default withEnglishFallback({
onDelete: "On delete",
onUpdate: "On update",
defaultAction: "Predefinito",
transwarpPartitionColumns: "Colonne di partizione",
transwarpBucketColumns: "Colonne di distribuzione",
transwarpBucketCount: "Numero di bucket",
transwarpStorageFormat: "Formato di archiviazione",
transwarpTransactional: "Tabella transazionale",
triggerName: "Trigger",
triggerStatement: "Istruzione trigger",
primary: "Primario",
+5
View File
@@ -4759,6 +4759,11 @@ export default withEnglishFallback({
onDelete: "削除時",
onUpdate: "更新時",
defaultAction: "デフォルト",
transwarpPartitionColumns: "パーティション列",
transwarpBucketColumns: "バケット列",
transwarpBucketCount: "バケット数",
transwarpStorageFormat: "保存形式",
transwarpTransactional: "トランザクションテーブル",
triggerName: "トリガー",
triggerStatement: "トリガー文",
primary: "主キー",
+5
View File
@@ -4694,6 +4694,11 @@ export default withEnglishFallback({
onDelete: "삭제 시",
onUpdate: "수정 시",
defaultAction: "기본",
transwarpPartitionColumns: "파티션 열",
transwarpBucketColumns: "버킷 열",
transwarpBucketCount: "버킷 수",
transwarpStorageFormat: "저장 형식",
transwarpTransactional: "트랜잭션 테이블",
triggerName: "트리거",
triggerStatement: "트리거 구문",
primary: "기본",
+5
View File
@@ -4697,6 +4697,11 @@ export default withEnglishFallback({
onDelete: "Ao excluir",
onUpdate: "Ao atualizar",
defaultAction: "Padrão",
transwarpPartitionColumns: "Colunas de partição",
transwarpBucketColumns: "Colunas de distribuição",
transwarpBucketCount: "Número de buckets",
transwarpStorageFormat: "Formato de armazenamento",
transwarpTransactional: "Tabela transacional",
triggerName: "Trigger",
triggerStatement: "Comando do trigger",
primary: "Primária",
+5
View File
@@ -5223,6 +5223,11 @@ export default withEnglishFallback({
onDelete: "При удалении",
onUpdate: "При обновлении",
defaultAction: "По умолчанию",
transwarpPartitionColumns: "Столбцы секционирования",
transwarpBucketColumns: "Столбцы распределения",
transwarpBucketCount: "Число корзин",
transwarpStorageFormat: "Формат хранения",
transwarpTransactional: "Транзакционная таблица",
triggerName: "Триггер",
triggerStatement: "Оператор триггера",
primary: "Первичный",
+5
View File
@@ -5251,6 +5251,11 @@ export default withEnglishFallback({
mysqlTableEngineSearchPlaceholder: "搜索存储引擎",
mysqlTableEngineEmpty: "没有可用的存储引擎",
mysqlTableEngineLoadFailed: "加载存储引擎失败:{message}",
transwarpPartitionColumns: "分区列",
transwarpBucketColumns: "分桶列",
transwarpBucketCount: "桶数",
transwarpStorageFormat: "存储格式",
transwarpTransactional: "事务表",
owner: "所有者",
ownerPlaceholder: "输入角色名...",
ownerSearchPlaceholder: "搜索或输入角色名...",
+5
View File
@@ -4432,6 +4432,11 @@ export default withEnglishFallback({
onDelete: "刪除時",
onUpdate: "更新時",
defaultAction: "預設",
transwarpPartitionColumns: "分區欄位",
transwarpBucketColumns: "分桶欄位",
transwarpBucketCount: "桶數",
transwarpStorageFormat: "儲存格式",
transwarpTransactional: "交易資料表",
triggerName: "觸發器名稱",
triggerStatement: "觸發語句",
primary: "主鍵",
@@ -0,0 +1,25 @@
import { describe, expect, it } from "vitest";
import { parseConnectionUrl } from "@/lib/connection/connectionUrl";
describe("Transwarp JDBC URL parsing", () => {
it("keeps the selected product when the SDK uses its other compatible URL", () => {
const inceptor = parseConnectionUrl("jdbc:inceptor2://quark:10000/default;fetchSize=200", "transwarp-inceptor");
expect(inceptor).toMatchObject({
dbType: "transwarp",
driverProfile: "transwarp-inceptor",
host: "quark",
port: 10000,
database: "default",
urlParams: "fetchSize=200",
});
expect(parseConnectionUrl("jdbc:transwarp2://quark:10000/default").driverProfile).toBe("transwarp-inceptor");
});
it("detects Inceptor URLs without an existing product selection", () => {
expect(parseConnectionUrl("jdbc:inceptor2://quark:10000/default")).toMatchObject({
dbType: "transwarp",
driverProfile: "transwarp-inceptor",
});
expect(parseConnectionUrl("jdbc:hive2://quark:10000/default").dbType).toBe("hive");
});
});
@@ -59,6 +59,11 @@ describe("connectionUrlCopyFormats", () => {
const formats = connectionUrlCopyFormats(config({ db_type: "hive", host: "hive.example.com", port: 10000, database: "dw" }));
expect(formats).toEqual(["jdbcUrl", "jdbcUrlWithCredentials", "hostPort"]);
});
it("uses semicolon options when copying an Inceptor JDBC URL", () => {
const transwarp = config({ db_type: "transwarp", host: "quark.example.com", port: 10000, database: "default", username: "", password: "", url_params: "fetchSize=500" });
expect(buildConnectionUrlCopy(transwarp, "jdbcUrl")).toBe("jdbc:inceptor2://quark.example.com:10000/default;fetchSize=500");
});
});
describe("buildConnectionUrlCopy standard URL", () => {
@@ -33,4 +33,11 @@ describe("database namespace creation targets", () => {
expect(connectionNamespaceCreationTarget(postgres)).toBe("database");
expect(databaseNodeNamespaceCreationTarget(postgres, databaseNode)).toBe("schema");
});
it("offers top-level database creation for Inceptor without nested schemas", () => {
const transwarp = { db_type: "transwarp", driver_profile: "transwarp-inceptor" } as Conn;
expect(connectionNamespaceCreationTarget(transwarp)).toBe("database");
expect(databaseNodeNamespaceCreationTarget(transwarp, databaseNode)).toBeNull();
});
});
@@ -667,6 +667,7 @@ describe("splitSqlStatementRanges", () => {
expect(rangeSqlTexts(splitSqlStatementRanges(argoProcedureFixture, "argo"))).toEqual([argoProcedureFixture]);
expect(hasMultipleExecutionTargets(argoProcedureFixture, "argo")).toBe(false);
expect(rangeSqlTexts(executableStatementRanges(argoProcedureFixture, "argo"))).toEqual([argoProcedureFixture]);
expect(rangeSqlTexts(splitSqlStatementRanges(argoProcedureFixture, "transwarp"))).toEqual([argoProcedureFixture]);
});
it("statement at cursor inside an ArgoDB procedure body returns the whole definition", () => {
@@ -1,6 +1,16 @@
import { describe, expect, it } from "vitest";
import { databaseRenameMaintenanceDatabase, supportsDatabaseRename } from "@/lib/table/objectRenameSql";
import { databaseRenameMaintenanceDatabase, supportsDatabaseRename, supportsObjectRename } from "@/lib/table/objectRenameSql";
describe("Inceptor object rename", () => {
it("exposes table rename without enabling unsupported object or database renames", () => {
expect(supportsObjectRename("transwarp", "TABLE")).toBe(true);
expect(supportsObjectRename("transwarp", "VIEW")).toBe(false);
expect(supportsObjectRename("transwarp", "PROCEDURE")).toBe(false);
expect(supportsObjectRename("transwarp", "FUNCTION")).toBe(false);
expect(supportsDatabaseRename("transwarp")).toBe(false);
});
});
describe("database rename", () => {
it("chooses a maintenance database outside the rename target", () => {
@@ -10,6 +10,7 @@ import {
editablePrimaryKeys,
editableRowIdentifierColumns,
hasCompleteTdengineRowIdentity,
hiveTablePropertiesIndicateTransactional,
isClickHouseExistingRowReadonlyColumn,
isHiddenGridColumn,
isSalesforceExistingRowReadonlyColumn,
@@ -45,6 +46,17 @@ function index(columns: string[], isUnique = true, filter: string | null = null)
}
describe("tableEditing", () => {
it("allows Inceptor inserts but only edits existing rows in transactional tables", () => {
expect(isTableDataEditable("transwarp", [], "TABLE")).toBe(true);
expect(isTableDataEditable("transwarp", [], "VIEW")).toBe(false);
expect(canInsertTableRows("transwarp")).toBe(true);
expect(canEditExistingTableRows("transwarp", false)).toBe(false);
expect(canEditExistingTableRows("transwarp", true)).toBe(true);
expect(supportsDataGridTransaction("transwarp")).toBe(true);
expect(hiveTablePropertiesIndicateTransactional({ rows: [["transactional", "true"]] })).toBe(true);
expect(hiveTablePropertiesIndicateTransactional({ rows: [["true"]] })).toBe(true);
expect(hiveTablePropertiesIndicateTransactional({ rows: [["false"]] })).toBe(false);
});
it("synthesizes ROWID only for Oracle-compatible base tables", () => {
expect(editablePrimaryKeys("oracle", [column("ID"), column("NAME")])).toEqual([]);
expect(editablePrimaryKeys("oracle", [column("ID"), column("NAME")], "VIEW")).toEqual([]);
@@ -0,0 +1,39 @@
import { describe, expect, it } from "vitest";
import { buildInceptorCreateOptions, emptyTablePhysicalOptions, hasTablePhysicalOptions, pruneTablePhysicalOptions, restoreTablePhysicalOptions } from "@/lib/table/tablePhysicalOptions";
const columns = [
{ id: "first", name: "id" },
{ id: "second", name: "day" },
{ id: "third", name: "label" },
];
describe("table physical options", () => {
it("restores a saved draft without sharing its column selections", () => {
const saved = { physicalOptions: { partitionColumnIds: ["second"], distributionColumnIds: ["first"], bucketCount: "2", storageFormat: "ORC", transactional: true } };
const restored = restoreTablePhysicalOptions(saved);
expect(restored).toMatchObject({ partitionColumnIds: ["second"], distributionColumnIds: ["first"], bucketCount: "2", storageFormat: "ORC", transactional: true });
expect(restored.partitionColumnIds).not.toBe(saved.physicalOptions.partitionColumnIds);
expect(restored.distributionColumnIds).not.toBe(saved.physicalOptions.distributionColumnIds);
expect(restoreTablePhysicalOptions({})).toEqual(emptyTablePhysicalOptions());
expect(hasTablePhysicalOptions(restored)).toBe(true);
expect(hasTablePhysicalOptions(emptyTablePhysicalOptions())).toBe(false);
});
it("uses the current column name after a selected column is renamed", () => {
const draft = { ...emptyTablePhysicalOptions(), partitionColumnIds: ["second"], distributionColumnIds: ["first"], bucketCount: "2", storageFormat: "ORC", transactional: true };
const renamed = columns.map((column) => (column.id === "second" ? { ...column, name: "event_day" } : column));
expect(buildInceptorCreateOptions(draft, renamed)).toEqual({
partitionColumns: ["event_day"],
bucketColumns: ["id"],
bucketCount: 2,
storageFormat: "ORC",
transactional: true,
});
expect(pruneTablePhysicalOptions(draft, new Set(["first", "third"]))).toMatchObject({ partitionColumnIds: [], distributionColumnIds: ["first"] });
});
it("keeps an invalid bucket count in the SQL builder's validation path", () => {
const draft = { ...emptyTablePhysicalOptions(), distributionColumnIds: ["first"], bucketCount: "1.5" };
expect(buildInceptorCreateOptions(draft, columns).bucketCount).toBe(0);
});
});
@@ -110,6 +110,21 @@ describe("tableStructureCapabilities", () => {
}
});
it("enables only the verified Inceptor column structure operations", () => {
expect(getTableStructureCapabilities("transwarp", "transwarp")).toMatchObject({
dialect: "mysql",
alterStrategy: "direct",
createTable: true,
addColumn: true,
dropColumn: false,
renameColumn: true,
alterExistingColumn: true,
comment: true,
createIndex: false,
foreignKey: false,
});
});
it("uses local-only column reordering for editable databases without physical reorder support", () => {
for (const databaseType of ["sqlserver", "postgres", "sqlite", "oracle", "dameng", "duckdb", "informix"] as const) {
expect(supportsLocalTableColumnReorder(databaseType, databaseType)).toBe(true);
@@ -28,6 +28,16 @@ describe("HiveServer2-compatible driver installation", () => {
});
});
describe("Transwarp driver installation", () => {
it("routes the single Inceptor profile to the shared Agent package", () => {
const profile = "transwarp-inceptor";
expect(agentDriverInstallKey("transwarp", profile)).toBe("transwarp");
expect(showAgentDriverInstallHint("transwarp", [], profile)).toBe(true);
expect(showAgentDriverInstallHint("transwarp", [{ db_type: "transwarp", installed: true }], profile)).toBe(false);
expect(driverStoreFocusForInstallError("transwarp driver is not installed. Please install it from the Driver Manager.", "transwarp", profile)).toEqual({ target: "driver", driver: "transwarp" });
});
});
describe("shouldApplyDriverStoreFocus", () => {
it("applies when the driver first appears after a list load", () => {
expect(shouldApplyDriverStoreFocus(null, "driver:mysql", false)).toBe(true);
@@ -110,6 +110,7 @@ describe("AGENT_DRIVER_CATEGORY_MAP integrity", () => {
"oceanbase-oracle",
"oscar",
"sundb",
"transwarp",
"uxdb",
"vastbase",
"xugu",
@@ -37,6 +37,7 @@ export function agentDriverInstallKey(dbType: DatabaseType | undefined, driverPr
if (dbType === "kyuubi" || dbType === "impala") return "hive";
if (dbType === "oracle") return "oracle";
if (dbType === "h2") return "h2";
if (dbType === "transwarp") return "transwarp";
if (dbType === "mongodb") return "mongodb";
if (dbType === "dameng") return "dameng";
if (dbType === "gbase") return driverProfile === "gbase8s" ? "gbase8s" : "gbase8a";
@@ -35,6 +35,7 @@ const DRIVER_STARTUP_FLOOR_TYPES = new Set<DatabaseType>([
"kyuubi",
"impala",
"argo",
"transwarp",
"spark",
"db2",
"informix",
@@ -265,6 +265,9 @@ export function connectionUrlPlaceholder(dbType: DatabaseType, driverProfile?: s
case "iris":
return driverProfile === "cache" ? "cache://user:password@host:port/namespace" : "iris://user:password@host:port/namespace";
case "transwarp":
return "jdbc:inceptor2://host:10000/default";
case "influxdb":
return "influxdb://user:password@host:port/database";
@@ -424,8 +424,8 @@ export function connectionProfileForScheme(scheme: string, preferredProfile?: st
return SCHEME_PROFILES[normalizedScheme];
}
function parseJdbcHiveUrl(source: string): ParsedConnectionUrl | null {
const match = /^jdbc:hive2:\/\/(?<hosts>[^/?#;]+)(?:\/(?<path>[^?#]*))?(?<query>\?[^#]*)?(?<fragment>#.*)?$/i.exec(source);
function parseJdbcHiveUrl(source: string, preferredProfile?: string): ParsedConnectionUrl | null {
const match = /^jdbc:(?<subprotocol>hive2|inceptor2|transwarp2):\/\/(?<hosts>[^/?#;]+)(?:\/(?<path>[^?#]*))?(?<query>\?[^#]*)?(?<fragment>#.*)?$/i.exec(source);
if (!match?.groups) return null;
const firstHost = match.groups.hosts.split(",")[0]?.trim();
@@ -442,11 +442,13 @@ function parseJdbcHiveUrl(source: string): ParsedConnectionUrl | null {
const [rawDatabase = "", ...paramParts] = (match.groups.path || "").split(";");
const structured = extractHiveStructuredParams(paramParts.join(";"));
const urlParams = `${structured.urlParams}${match.groups.query || ""}${match.groups.fragment || ""}`;
const profile = preferredProfile === "transwarp-inceptor" ? preferredProfile : match.groups.subprotocol.toLowerCase() === "hive2" ? "hive" : "transwarp-inceptor";
const transwarp = profile !== "hive";
return {
dbType: "hive",
driverProfile: "hive",
driverLabel: "Apache Hive",
dbType: transwarp ? "transwarp" : "hive",
driverProfile: profile,
driverLabel: profile === "transwarp-inceptor" ? "星环Inceptor" : "Apache Hive",
host: endpoint.hostname.replace(/^\[(.*)]$/, "$1"),
port: endpoint.port ? Number(endpoint.port) : 10000,
username: structured.username ?? decodeUrlPart(endpoint.username),
@@ -698,7 +700,7 @@ export function parseConnectionUrl(value: string, preferredProfile?: string): Pa
if (/^jdbc:oceanbase:(?:oracle:)?loadbalance:\/\//i.test(input)) {
throw new Error("Unsupported OceanBase JDBC URL variant: loadbalance");
}
const jdbcHive = parseJdbcHiveUrl(input);
const jdbcHive = parseJdbcHiveUrl(input, preferredProfile);
if (jdbcHive) return jdbcHive;
const jdbcH2 = parseH2JdbcUrl(input);
if (jdbcH2) return jdbcH2;
@@ -100,6 +100,7 @@ const JDBC_URL_PREFIXES: Partial<Record<DatabaseType, string>> = {
hive: "jdbc:hive2",
kyuubi: "jdbc:hive2",
argo: "jdbc:hive2",
transwarp: "jdbc:inceptor2",
impala: "jdbc:impala",
trino: "jdbc:trino",
prestosql: "jdbc:presto",
@@ -353,6 +354,12 @@ function buildSpecialJdbcUrl(config: ConnectionUrlCopyConfig, database: string,
const query = joinNonEmpty("&", [database && `db=${database}`, ...credentials, rawParams.replace(/^[?&]+/, "")]);
return `jdbc:snowflake://${formatHostForUrl(host)}/${query ? `?${query}` : ""}`;
}
case "transwarp": {
const portPart = shouldAppendPort(config) ? `:${config.port}` : "";
const props = rawParams.replace(/^[?&;]+/, "").replace(/&/g, ";");
const ssl = config.ssl && !queryHasParam(props, ["ssl"]) ? ";ssl=true" : "";
return `jdbc:inceptor2://${formatHostForUrl(host)}${portPart}/${database || "default"}${props ? `;${props}` : ""}${ssl}`;
}
case "informix":
case "gbase": {
const prefix = config.db_type === "informix" ? "jdbc:informix-sqli" : jdbcPrefixFor(config);
@@ -61,6 +61,7 @@ export const AGENT_DRIVER_CATEGORY_MAP: Readonly<Record<string, DriverCategoryKe
sundb: "domestic",
tdengine: "timeseries",
teradata: "analytics",
transwarp: "domestic",
trino: "analytics",
uxdb: "domestic",
vastbase: "domestic",
@@ -134,6 +134,7 @@ export interface DataGridConditionalUpdateSqlOptions {
}
export interface HiveTablePropertiesSqlOptions {
databaseType?: DatabaseType;
schema?: string;
tableName: string;
propertyName: string;
@@ -4,7 +4,7 @@ import { manifestDatabaseTypes } from "@/lib/database/databaseDriverManifest";
import { supportsTransfer } from "@/lib/database/databaseFeatureSupport";
import type { DatabaseType } from "@/types/database";
const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "turso", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "highgo", "vastbase", "goldendb", "yashandb", "questdb", "h2", "hive", "argo", "kyuubi", "impala", "spark"];
const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "turso", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "highgo", "vastbase", "goldendb", "yashandb", "questdb", "h2", "hive", "argo", "transwarp", "kyuubi", "impala", "spark"];
describe("transferObjectKinds", () => {
it("groups databases into transfer families", () => {
@@ -70,6 +70,7 @@ export const DATABASE_NAMESPACE_CREATION_MATRIX = {
prestosql: { database: "schema" },
hive: { deferred: "Hive database creation needs agent metadata validation first" },
argo: { deferred: "ArgoDB database creation needs agent metadata validation first" },
transwarp: { connection: "database" },
kyuubi: { deferred: "Kyuubi database creation needs dedicated metadata validation first" },
impala: { deferred: "Impala database creation needs dedicated metadata validation first" },
spark: { deferred: "Spark database creation needs agent metadata validation first" },
@@ -80,6 +80,7 @@ const DATABASE_TYPE_OBJECTS = new Map<DatabaseType, SidebarObjectKind[]>([
// ArgoDB (Transwarp) shares the Hive agent; its catalog views
// (system.procedures_v / system.functions_v) expose routines natively.
["argo", ROUTINE_OBJECTS],
["transwarp", ["TABLE", "VIEW", "PROCEDURE", "FUNCTION", "PACKAGE", "PACKAGE_BODY"]],
["kyuubi", TABLE_VIEW_OBJECTS],
["impala", TABLE_VIEW_OBJECTS],
["spark", TABLE_VIEW_OBJECTS],
@@ -71,6 +71,7 @@ export const DATABASE_PROPERTY_EDITING_MATRIX = {
prestosql: { deferred: "catalog properties are connector-managed" },
hive: { deferred: "database properties need agent metadata validation first" },
argo: { deferred: "database properties need agent metadata validation first" },
transwarp: { deferred: "database properties need server validation first" },
kyuubi: { deferred: "database properties need Kyuubi-specific validation first" },
impala: { deferred: "database properties need Impala-specific validation first" },
spark: { deferred: "database properties need agent metadata validation first" },
@@ -133,6 +133,16 @@ const DATABASE_CAPABILITY_OVERRIDES: Partial<Record<DatabaseType, Partial<Databa
transaction: false,
},
},
transwarp: {
tableData: {
insert: true,
updateRequiresPrimaryKey: false,
deleteRequiresPrimaryKey: false,
keylessRowPredicate: true,
requiresTransactionalTableForExistingRows: true,
transaction: true,
},
},
jdbc: {
tableData: {
insert: false,
+1 -1
View File
@@ -15,7 +15,7 @@ const GAUSSDB_COUNT_QUERY_DOP_KEY = "gaussdbCountQueryDop";
export const GAUSSDB_M_JDBC_DRIVER_PROFILE = "gaussdb-m";
export const GAUSSDB_M_JDBC_DRIVER_CLASS = "com.huawei.gaussdb.jdbc.Driver";
const DATABASE_AS_EXECUTION_SCHEMA_TYPES = new Set<DatabaseType>(["hive", "kyuubi", "impala", "argo", "spark"]);
const DATABASE_AS_EXECUTION_SCHEMA_TYPES = new Set<DatabaseType>(["hive", "kyuubi", "impala", "argo", "transwarp", "spark"]);
const CONNECTION_ROOT_SCHEMA_TYPES = new Set<DatabaseType>(["oracle", "dameng", "oceanbase-oracle"]);
const JDBC_DIALECT_MATCHERS: Array<{ type: DatabaseType; patterns: RegExp[] }> = [
@@ -14,7 +14,7 @@ const SQLITE_CODEMIRROR_DATABASE_TYPES = new Set<DatabaseType>(["sqlite", "rqlit
// literals; the mysql-family types are covered by isMysql at the define() site. Mirrors
// BACKSLASH_ESCAPE_STRING_DIALECTS in lib/sql/sqlStatementRanges.ts so the editor
// tokenizer and the statement splitter agree on escape semantics per dialect.
const BACKSLASH_ESCAPE_CODEMIRROR_DATABASE_TYPES = new Set<DatabaseType>(["hive", "argo", "impala", "spark", "databend"]);
const BACKSLASH_ESCAPE_CODEMIRROR_DATABASE_TYPES = new Set<DatabaseType>(["hive", "argo", "transwarp", "impala", "spark", "databend"]);
const CODEMIRROR_SQLITE_EXTENSION_KEYWORDS = new Set("abort analyze attach autoincrement conflict database detach exclusive fail glob ignore index indexed instead isnull notnull offset plan pragma query raise regexp reindex rename replace temp vacuum virtual".split(" "));
const STANDARD_SQL_TYPES = "array binary bit boolean char character clob date decimal double float int integer interval large national nchar nclob numeric object precision real smallint time timestamp varchar varying";
@@ -4,7 +4,7 @@ import { qualifiedTableName, quoteTableIdentifier } from "@/lib/table/tableSelec
import { requiresMysqlIdentifierQuote, requiresPostgresIdentifierQuote } from "@/lib/sql/sqlIdentifier.ts";
/** 智能引号使用反引号的方言族。 */
const SMART_QUOTE_BACKTICK_TYPES = new Set<DatabaseType>(["mysql", "clickhouse", "hive", "argo", "kyuubi", "impala", "spark", "databricks", "databend", "tdengine", "access", "doris", "starrocks", "goldendb"]);
const SMART_QUOTE_BACKTICK_TYPES = new Set<DatabaseType>(["mysql", "clickhouse", "hive", "argo", "transwarp", "kyuubi", "impala", "spark", "databricks", "databend", "tdengine", "access", "doris", "starrocks", "goldendb"]);
/** 智能引号使用双引号的方言族(SQL Server 方括号除外)。 */
const SMART_QUOTE_DOUBLE_TYPES = new Set<DatabaseType>(["postgres", "gaussdb", "opengauss"]);
/** SQL Server 族智能引号使用方括号。 */
+2 -2
View File
@@ -1439,7 +1439,7 @@ type SqlCompletionApplyDialect = "mysql" | "postgres" | "sqlserver" | "oracle" |
// QueryEditor may use another dialect as a CodeMirror syntax fallback.
// Completion apply text must still follow the connected database's identifier
// folding and quoting rules.
const MYSQL_LIKE_IDENTIFIER_DATABASES = new Set<DatabaseType>(["mysql", "clickhouse", "hive", "argo", "kyuubi", "impala", "spark", "databend", "tdengine", "access", "doris", "starrocks"]);
const MYSQL_LIKE_IDENTIFIER_DATABASES = new Set<DatabaseType>(["mysql", "clickhouse", "hive", "argo", "transwarp", "kyuubi", "impala", "spark", "databend", "tdengine", "access", "doris", "starrocks"]);
const POSTGRES_LIKE_IDENTIFIER_DATABASES = new Set<DatabaseType>(["postgres", "redshift", "gaussdb", "kingbase", "highgo", "uxdb", "vastbase", "kwdb", "opengauss"]);
export const ORACLE_COMPAT_IDENTIFIER_DATABASES = new Set<DatabaseType>(["oracle", "oceanbase-oracle", "yashandb", "oscar", "xugu"]);
const UPPER_FOLDING_IDENTIFIER_DATABASES = new Set<DatabaseType>(["dameng", "db2"]);
@@ -3256,7 +3256,7 @@ function lastTopLevelKeywordIndex(sql: string, keyword: string): number {
// unquoted token as a table reference there is a false-positive risk. Everything else defaults
// to Unicode support (MySQL/Postgres/SQL Server/SQLite/DuckDB and the many Chinese-vendor
// engines this product supports legitimately use non-ASCII unquoted identifiers).
const ASCII_ONLY_UNQUOTED_IDENTIFIER_DATABASES = new Set<DatabaseType>(["clickhouse", "snowflake", "bigquery", "hive", "argo", "spark", "trino", "prestosql", "impala", "db2", "teradata"]);
const ASCII_ONLY_UNQUOTED_IDENTIFIER_DATABASES = new Set<DatabaseType>(["clickhouse", "snowflake", "bigquery", "hive", "argo", "transwarp", "spark", "trino", "prestosql", "impala", "db2", "teradata"]);
// Dialects where a bare "--" needs trailing whitespace/EOL to start a comment, since MySQL
// reserves unspaced "--" for double-negation (e.g. `SELECT 1--1`). Scoped to mysql itself plus
@@ -293,7 +293,7 @@ const SET_OPERATION_MODIFIER_KEYWORDS = new Set(["ALL", "DISTINCT"]);
// in sync. ArgoDB (Transwarp Hive/Inceptor fork) ships a PL/SQL-compatible procedure
// language (`CREATE [OR REPLACE] PROCEDURE ... IS BEGIN ... END;`), so its statement
// ranges must stay whole instead of splitting at every body semicolon.
const ORACLE_LIKE_PL_SQL_DATABASES: ReadonlySet<DatabaseType> = new Set(["oracle", "dameng", "gaussdb", "yashandb", "oscar", "oceanbase-oracle", "xugu", "argo"]);
const ORACLE_LIKE_PL_SQL_DATABASES: ReadonlySet<DatabaseType> = new Set(["oracle", "dameng", "gaussdb", "yashandb", "oscar", "oceanbase-oracle", "xugu", "argo", "transwarp"]);
const MYSQL_ROUTINE_BLOCK_DATABASES: ReadonlySet<DatabaseType> = new Set(["mysql", "doris", "starrocks", "manticoresearch", "goldendb"]);
// PostgreSQL/openGauss are also the connection types users pick for GaussDB/openGauss instances
// running in Oracle (A) compatibility mode, where a routine body is written in Oracle style
@@ -305,7 +305,7 @@ const POSTGRES_FAMILY_DATABASES: ReadonlySet<DatabaseType> = new Set(["postgres"
// escape unconditionally makes ESCAPE '\' swallow its closing quote and the following statement
// boundary, so the next statement loses its run button (#8189). Gate it by dialect, matching the
// tokenizer/completion side.
export const BACKSLASH_ESCAPE_STRING_DIALECTS: ReadonlySet<DatabaseType> = new Set(["mysql", "doris", "starrocks", "hive", "argo", "impala", "spark", "databend"]);
export const BACKSLASH_ESCAPE_STRING_DIALECTS: ReadonlySet<DatabaseType> = new Set(["mysql", "doris", "starrocks", "hive", "argo", "transwarp", "impala", "spark", "databend"]);
function allowsBackslashStringEscape(databaseType?: DatabaseType): boolean {
return !!databaseType && BACKSLASH_ESCAPE_STRING_DIALECTS.has(databaseType);
}
@@ -23,6 +23,7 @@ export function supportsObjectRename(databaseType: DatabaseType | undefined, obj
return false;
}
if (databaseType === "sqlite" || databaseType === "rqlite" || databaseType === "turso" || databaseType === "cloudflare-d1" || databaseType === "duckdb") return objectType === "TABLE";
if (databaseType === "transwarp") return objectType === "TABLE";
if (databaseType === "mysql" || databaseType === "goldendb") return objectType === "TABLE" || objectType === "VIEW";
if (postgresLikeRenameTypes.has(databaseType)) return objectType === "TABLE" || objectType === "VIEW" || objectType === "MATERIALIZED_VIEW";
if (oracleLikeRenameTypes.has(databaseType)) return objectType === "TABLE" || objectType === "VIEW" || objectType === "MATERIALIZED_VIEW";
@@ -92,6 +92,12 @@ export function canDeleteExistingTdengineRows(databaseType: DatabaseType | undef
export function hiveTablePropertiesIndicateTransactional(result: { rows: readonly (readonly unknown[])[] }): boolean {
return result.rows.some((row) => {
if (row.length === 1)
return (
String(row[0] ?? "")
.trim()
.toLowerCase() === "true"
);
const name = String(row[0] ?? "")
.trim()
.toLowerCase();
@@ -0,0 +1,68 @@
import type { EditableStructureColumn, TranswarpCreateTableOptions } from "@/lib/table/tableStructureEditorSql";
export interface TablePhysicalOptionsDraft {
partitionColumnIds: string[];
distributionColumnIds: string[];
bucketCount: string;
storageFormat: string;
transactional: boolean;
}
export interface TablePhysicalOptionsConfig {
partitionColumnsLabel: string;
distributionColumnsLabel: string;
bucketCountLabel: string;
storageFormatLabel: string;
transactionalLabel?: string;
maxBuckets: number;
storageFormats: readonly string[];
}
export const INCEPTOR_PHYSICAL_OPTIONS: TablePhysicalOptionsConfig = {
partitionColumnsLabel: "structureEditor.transwarpPartitionColumns",
distributionColumnsLabel: "structureEditor.transwarpBucketColumns",
bucketCountLabel: "structureEditor.transwarpBucketCount",
storageFormatLabel: "structureEditor.transwarpStorageFormat",
transactionalLabel: "structureEditor.transwarpTransactional",
maxBuckets: 4096,
storageFormats: ["ORC", "PARQUET", "TEXTFILE"],
};
export function emptyTablePhysicalOptions(): TablePhysicalOptionsDraft {
return { partitionColumnIds: [], distributionColumnIds: [], bucketCount: "", storageFormat: "", transactional: false };
}
export function hasTablePhysicalOptions(draft: TablePhysicalOptionsDraft): boolean {
return draft.partitionColumnIds.length > 0 || draft.distributionColumnIds.length > 0 || !!draft.bucketCount.trim() || !!draft.storageFormat || draft.transactional;
}
export function pruneTablePhysicalOptions(draft: TablePhysicalOptionsDraft, availableColumnIds: ReadonlySet<string>): TablePhysicalOptionsDraft {
const partitionColumnIds = draft.partitionColumnIds.filter((id) => availableColumnIds.has(id));
const distributionColumnIds = draft.distributionColumnIds.filter((id) => availableColumnIds.has(id));
if (partitionColumnIds.length === draft.partitionColumnIds.length && distributionColumnIds.length === draft.distributionColumnIds.length) return draft;
return { ...draft, partitionColumnIds, distributionColumnIds };
}
export function restoreTablePhysicalOptions(saved: { physicalOptions?: TablePhysicalOptionsDraft }): TablePhysicalOptionsDraft {
if (!saved.physicalOptions) return emptyTablePhysicalOptions();
return {
partitionColumnIds: [...saved.physicalOptions.partitionColumnIds],
distributionColumnIds: [...saved.physicalOptions.distributionColumnIds],
bucketCount: saved.physicalOptions.bucketCount,
storageFormat: saved.physicalOptions.storageFormat,
transactional: saved.physicalOptions.transactional,
};
}
export function buildInceptorCreateOptions(draft: TablePhysicalOptionsDraft, columns: readonly Pick<EditableStructureColumn, "id" | "name">[]): TranswarpCreateTableOptions {
const names = (ids: readonly string[]) => ids.map((id) => columns.find((column) => column.id === id)?.name ?? id);
const count = draft.bucketCount.trim();
const parsedCount = Number(count);
return {
partitionColumns: names(draft.partitionColumnIds),
bucketColumns: names(draft.distributionColumnIds),
bucketCount: count ? (Number.isSafeInteger(parsedCount) && parsedCount > 0 && parsedCount <= 4096 ? parsedCount : 0) : undefined,
storageFormat: draft.storageFormat || undefined,
transactional: draft.transactional,
};
}
@@ -109,6 +109,7 @@ export function quoteTableIdentifier(databaseType: DatabaseType | undefined, nam
databaseType === "clickhouse" ||
databaseType === "hive" ||
databaseType === "argo" ||
databaseType === "transwarp" ||
databaseType === "kyuubi" ||
databaseType === "impala" ||
databaseType === "spark" ||
@@ -152,6 +153,7 @@ function requiresIdentifierQuote(databaseType: DatabaseType | undefined, name: s
case "clickhouse":
case "hive":
case "argo":
case "transwarp":
case "kyuubi":
case "impala":
case "spark":
@@ -268,6 +268,20 @@ const informixCapabilities = capabilities({
rebuildIndex: true,
});
// Inceptor accepts ADD COLUMNS and CHANGE. DROP COLUMN and index/constraint DDL
// remain disabled until a safe server-supported form is verified.
const transwarpCapabilities = capabilities({
dialect: "mysql",
createTable: true,
addColumn: true,
renameColumn: true,
alterExistingColumn: true,
alterType: true,
alterNullability: true,
alterDefault: true,
comment: true,
});
const accessCapabilities = capabilities({
dialect: "h2",
createTable: true,
@@ -359,6 +373,7 @@ const capabilityByType: Partial<Record<DatabaseType, TableStructureCapabilities>
access: accessCapabilities,
clickhouse: clickhouseCapabilities,
informix: informixCapabilities,
transwarp: transwarpCapabilities,
influxdb: influxdbCapabilities,
manticoresearch: manticoreSearchCapabilities,
};
@@ -90,6 +90,7 @@ export interface BuildTableStructureChangeSqlOptions {
tableComment?: string;
originalTableComment?: string;
mysqlEngine?: string;
transwarpCreate?: TranswarpCreateTableOptions;
/** MySQL only: the table's current default collation. Columns whose collation merely
* matches it inherit the table default, so the backend leaves their redundant
* `CHARACTER SET`/`COLLATE` clauses out of the generated DDL. */
@@ -103,6 +104,14 @@ export interface BuildTableStructureChangeSqlOptions {
isGaussdbMMode?: boolean;
}
export interface TranswarpCreateTableOptions {
partitionColumns: string[];
bucketColumns: string[];
bucketCount?: number;
storageFormat?: string;
transactional: boolean;
}
export interface TableStructureChangeSql {
statements: string[];
warnings: string[];
+1
View File
@@ -1782,6 +1782,7 @@ export interface TableStructureEditorDraft {
originalMysqlAutoIncrementValue?: string;
mysqlTableEngine?: string;
originalMysqlTableEngine?: string;
physicalOptions?: import("@/lib/table/tablePhysicalOptions").TablePhysicalOptionsDraft;
tableOwner?: string;
originalTableOwner?: string;
columns: import("@/lib/table/tableStructureEditorSql").EditableStructureColumn[];
@@ -87,6 +87,7 @@ export const CONNECTION_PROFILES = {
prestosql: { type: "prestosql", port: 8080, user: "", label: "PrestoSQL", icon: "presto" },
hive: { type: "hive", port: 10000, user: "", label: "Apache Hive", icon: "hive" },
kyuubi: { type: "kyuubi", port: 10009, user: "", label: "Apache Kyuubi", icon: "kyuubi", urlParams: "auth=NONE" },
"transwarp-inceptor": { type: "transwarp", port: 10000, user: "", label: "星环Inceptor", icon: "transwarp-inceptor" },
argo: { type: "argo", port: 10000, user: "", label: "ArgoDB (Transwarp)", icon: "hive", urlParams: "auth=noSasl" },
impala: { type: "impala", port: 21050, user: "", label: "Apache Impala", icon: "impala", urlParams: "auth=noSasl" },
spark: { type: "spark", port: 10015, user: "", label: "Apache Spark", icon: "spark" },
@@ -196,6 +197,7 @@ export const CONNECTION_PROFILE_ICONS = {
prestosql: "prestosql",
hive: "hive",
kyuubi: "kyuubi",
"transwarp-inceptor": "transwarp-inceptor",
argo: "hive",
impala: "impala",
spark: "spark",
@@ -300,6 +302,7 @@ export const CONNECTION_PICKER_OPTIONS = [
{ value: "prestosql", label: "PrestoSQL", category: "analytics" },
{ value: "hive", label: "Hive", category: "analytics" },
{ value: "kyuubi", label: "Apache Kyuubi", category: "analytics" },
{ value: "transwarp-inceptor", label: "星环Inceptor", category: "domestic" },
{ value: "argo", label: "ArgoDB", category: "domestic" },
{ value: "impala", label: "Apache Impala", category: "analytics" },
{ value: "spark", label: "Apache Spark", category: "analytics" },
@@ -53,6 +53,7 @@ export const DATABASE_TYPES = [
"prestosql",
"hive",
"argo",
"transwarp",
"kyuubi",
"impala",
"db2",
@@ -1906,6 +1906,50 @@
"driverManagement": true
}
},
{
"dbType": "transwarp",
"label": "星环Inceptor",
"dialect": "Transwarp",
"runtimeMode": "agent",
"mcpMode": "bridge",
"agentKey": "transwarp",
"driverStoreVisible": true,
"driverStoreOrder": 52,
"driverProfiles": [
{
"profile": "transwarp-inceptor",
"agentKey": "transwarp",
"label": "星环Inceptor",
"storeVisible": false
}
],
"singleConnectionPool": false,
"metadataConnectionScoped": false,
"skipTcpProbe": true,
"defaultPort": 10000,
"traits": {
"schemaAware": true
},
"supportLevel": "operate",
"capabilities": {
"queryExecution": true,
"metadataBrowse": true,
"objectBrowser": true,
"objectSource": true,
"schemaSearch": true,
"diagram": false,
"tableDataEdit": true,
"tableStructureEdit": true,
"tableImport": true,
"dataTransfer": true,
"sqlFileExecution": true,
"databaseCreate": true,
"fieldLineage": false,
"sqlExplain": false,
"userAdmin": false,
"driverManagement": true
}
},
{
"dbType": "kyuubi",
"label": "Apache Kyuubi",
+1
View File
@@ -325,6 +325,7 @@ macro_rules! agent_connection_pool_database_type {
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
+38 -1
View File
@@ -4891,7 +4891,7 @@ fn effective_import_batch_size(db_type: &DatabaseType, requested: usize) -> usiz
let max_rows = match db_type {
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_IMPORT_BATCH_ROWS,
DatabaseType::Iris => 1,
DatabaseType::CloudflareD1 => 100,
DatabaseType::CloudflareD1 | DatabaseType::Transwarp => 100,
DatabaseType::SqlServer => 1000,
DatabaseType::Sqlite => SQLITE_APPEND_COMMIT_ROWS,
_ => usize::MAX,
@@ -5204,6 +5204,7 @@ fn text_data_type(db_type: &DatabaseType) -> &'static str {
DatabaseType::Oracle | DatabaseType::OceanbaseOracle | DatabaseType::Dameng => "CLOB",
DatabaseType::ClickHouse => "String",
DatabaseType::Hive
| DatabaseType::Transwarp
| DatabaseType::Kyuubi
| DatabaseType::Trino
| DatabaseType::PrestoSql
@@ -12085,6 +12086,42 @@ mod tests {
]);
}
#[test]
fn transwarp_import_batches_preserve_values_across_multiple_statements() {
let mappings = vec!["id", "label", "payload"]
.into_iter()
.map(|name| TableImportColumnMapping {
source_column: name.to_string(),
target_column: name.to_string(),
target_data_type: None,
})
.collect::<Vec<_>>();
let data = ParsedImportFile {
columns: vec!["id".into(), "label".into(), "payload".into()],
rows: (0..205)
.map(|index| {
vec![
serde_json::json!(index),
if index == 101 { serde_json::Value::Null } else { serde_json::json!("dbx\u{4e2d}\u{6587}'s") },
serde_json::json!(r#"{"key":[1,2]}"#),
]
})
.collect(),
total_rows: 205,
effective_encoding: None,
};
let batches =
build_import_insert_batches(&data, &mappings, &[], "events", "analytics", &DatabaseType::Transwarp, 500)
.unwrap();
assert_eq!(batches.iter().map(|batch| batch.row_count).collect::<Vec<_>>(), vec![100, 100, 5]);
assert!(batches.iter().all(|batch| batch
.sql
.starts_with("INSERT INTO `analytics`.`events` (`id`, `label`, `payload`)\nSELECT ")));
assert!(batches[0].sql.contains("dbx\u{4e2d}\u{6587}''s"));
assert!(batches[1].sql.contains("SELECT 101, NULL"));
assert!(batches[2].sql.contains("SELECT 204,"));
}
#[test]
fn starrocks_csv_json_array_import_uses_typed_json_expression() {
let plan = CompiledImportPlan {
+232 -7
View File
@@ -121,6 +121,7 @@ impl SqlBatchLimits {
let max_rows = requested_max_rows.max(1).min(match db_type {
DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
DatabaseType::Oracle | DatabaseType::OceanbaseOracle => MAX_ORACLE_INSERT_ALL_ROWS,
DatabaseType::Transwarp => 100,
_ => usize::MAX,
});
let target_sql_bytes = match db_type {
@@ -3439,7 +3440,10 @@ pub fn map_column_type(source_type: &str, source_db: &DatabaseType, target_db: &
// Extract basic type, `bigint unsigned` -> `bigint`
base = base.split(' ').next().unwrap_or(base).trim();
if matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo) {
if matches!(
target_db,
DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo | DatabaseType::Transwarp
) {
return match base {
"tinyint" => "TINYINT".into(),
"smallint" | "int2" => "SMALLINT".into(),
@@ -3684,7 +3688,11 @@ fn generate_create_table_ddl_with_column_quoting(
if !c.is_nullable
&& !matches!(
target_db,
DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo
DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
)
{
line.push_str(" NOT NULL");
@@ -3709,7 +3717,10 @@ fn generate_create_table_ddl_with_column_quoting(
}
let mut pks = Vec::with_capacity(columns.iter().filter(|c| c.is_primary_key).count());
if !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo) {
if !matches!(
target_db,
DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo | DatabaseType::Transwarp
) {
for c in columns {
if c.is_primary_key {
let qname = transfer_column_identifier(&c.name, target_db, quote_target_column_names);
@@ -3881,6 +3892,7 @@ pub(crate) fn generate_insert_typed_from_value_rows(
struct InsertSqlTemplate {
standard_prefix: String,
oracle_into_prefix: Option<String>,
inceptor_select_prefix: Option<String>,
xugu_multirow_values: bool,
}
@@ -3927,6 +3939,8 @@ impl InsertSqlTemplate {
// round trip per row), which makes CSV imports orders of magnitude slower.
oracle_into_prefix: matches!(db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle)
.then(|| format!("INTO {full_table} ({col_list}) VALUES ")),
inceptor_select_prefix: (matches!(db_type, DatabaseType::Transwarp) && !columns.is_empty())
.then(|| format!("INSERT INTO {full_table} ({col_list})\n")),
// Xugu accepts consecutive row constructors (`VALUES (...) (...)`) but rejects
// the comma-separated multi-row form emitted by the generic template.
xugu_multirow_values: matches!(db_type, DatabaseType::Xugu),
@@ -3937,6 +3951,22 @@ impl InsertSqlTemplate {
if value_rows.is_empty() {
return String::new();
}
if let Some(prefix) = self.inceptor_select_prefix.as_deref() {
let mut sql = String::with_capacity(self.statement_bytes(
value_rows.iter().map(String::len).sum(),
value_rows.len(),
&DatabaseType::Transwarp,
));
sql.push_str(prefix);
for (index, values) in value_rows.iter().enumerate() {
if index > 0 {
sql.push_str("\nUNION ALL\n");
}
sql.push_str("SELECT ");
sql.push_str(values.strip_prefix('(').and_then(|row| row.strip_suffix(')')).unwrap_or(values));
}
return sql;
}
if let Some(into_prefix) = self.oracle_into_prefix.as_deref().filter(|_| value_rows.len() > 1) {
let capacity = "INSERT ALL\n".len()
+ into_prefix.len().saturating_mul(value_rows.len())
@@ -3975,6 +4005,12 @@ impl InsertSqlTemplate {
}
fn statement_bytes(&self, value_rows_bytes: usize, row_count: usize, db_type: &DatabaseType) -> usize {
if let Some(prefix) = self.inceptor_select_prefix.as_deref() {
return sql_text_bytes(prefix, db_type)
.saturating_add(value_rows_bytes.saturating_sub(2usize.saturating_mul(row_count)))
.saturating_add("SELECT ".len().saturating_mul(row_count))
.saturating_add("\nUNION ALL\n".len().saturating_mul(row_count.saturating_sub(1)));
}
if let Some(into_prefix) = self.oracle_into_prefix.as_deref().filter(|_| row_count > 1) {
return sql_text_bytes("INSERT ALL\n", db_type)
.saturating_add(sql_text_bytes(into_prefix, db_type).saturating_mul(row_count))
@@ -4399,7 +4435,14 @@ fn generate_upsert_typed_for_transfer(
fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize {
match (db_type, mode) {
(DatabaseType::SqlServer, TransferMode::Append | TransferMode::Overwrite) => MAX_SQLSERVER_INSERT_ROWS,
(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo, _) => 500,
(
DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp,
_,
) => 500,
(DatabaseType::Oracle, TransferMode::Append | TransferMode::Overwrite) => MAX_ORACLE_INSERT_ALL_ROWS,
// The INSERT ALL template is shared with OceanBase's Oracle mode, so its
// append/overwrite batches must clamp to the same per-statement row count.
@@ -8798,6 +8841,45 @@ struct HiveServerTransferCursor {
session_id: Option<String>,
}
fn is_transwarp_complex_type(data_type: &str) -> bool {
matches!(
data_type.trim().split('<').next().unwrap_or("").trim().to_ascii_lowercase().as_str(),
"array" | "map" | "struct" | "uniontype"
)
}
fn should_use_transwarp_server_side_copy(
source_db_type: &DatabaseType,
content: &TransferContent,
columns: &[db::ColumnInfo],
) -> bool {
*source_db_type == DatabaseType::Transwarp
&& should_copy_data(content)
&& columns.iter().any(|column| is_transwarp_complex_type(&column.data_type))
}
fn transwarp_server_side_copy_sql(
source_columns: &[String],
target_columns: &[String],
source_table: &str,
source_schema: &str,
target_table: &str,
target_schema: &str,
quote_target_columns: bool,
) -> String {
let db_type = DatabaseType::Transwarp;
let source = qualified_table(source_table, source_schema, &db_type, None);
let target = qualified_table(target_table, target_schema, &db_type, None);
let source_columns =
source_columns.iter().map(|name| quote_identifier(name, &db_type)).collect::<Vec<_>>().join(", ");
let target_columns = target_columns
.iter()
.map(|name| transfer_column_identifier(name, &db_type, quote_target_columns))
.collect::<Vec<_>>()
.join(", ");
format!("INSERT INTO {target} ({target_columns}) SELECT {source_columns} FROM {source}")
}
fn transfer_cursor_sql(
columns: &[String],
table: &str,
@@ -8826,7 +8908,7 @@ async fn fetch_hive_server_transfer_batch(
};
let pool_handle = state.pool_handle(pool_key).await;
let Some(PoolKind::Agent(client)) = pool_handle.as_ref() else {
return Err("Impala transfer requires an Agent connection".to_string());
return Err("Agent cursor transfer requires an Agent connection".to_string());
};
let client = client.clone();
@@ -9616,6 +9698,26 @@ where
};
log::info!("[transfer] {} total_rows={:?}", table, total_rows);
let server_side_complex_copy =
should_use_transwarp_server_side_copy(source_db_type, &request.content, &writable_columns);
if server_side_complex_copy {
if *target_db_type != DatabaseType::Transwarp || request.source_connection_id != request.target_connection_id {
return Err(format!(
"Native Inceptor complex columns in '{table}' require source and target on the same Transwarp connection"
));
}
let source_table =
qualified_table(table, &request.source_schema, source_db_type, request.source_catalog.as_deref());
let destination_table =
qualified_table(&target_table, &request.target_schema, target_db_type, request.target_catalog.as_deref());
if source_table.eq_ignore_ascii_case(&destination_table) {
return Err(format!("Cannot transfer native complex columns in '{table}' onto the same table"));
}
if total_rows.is_none() {
return Err(format!("Cannot count native complex rows in '{table}' before transfer"));
}
}
// Create table on target if requested
if request.create_table {
create_transfer_target_table(
@@ -9695,6 +9797,7 @@ where
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
))
|| matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng | DatabaseType::H2);
let target_columns = if needs_target_columns {
@@ -9761,6 +9864,27 @@ where
col_names.clone()
};
if server_side_complex_copy {
for (source_column, target_name) in writable_columns.iter().zip(&write_col_names) {
let Some(target_column) =
target_columns.iter().find(|column| column.name.eq_ignore_ascii_case(target_name))
else {
if target_table_preexisting {
return Err(format!("Target table '{target_table}' is missing column '{target_name}'"));
}
continue;
};
if is_transwarp_complex_type(&source_column.data_type)
&& !source_column.data_type.trim().eq_ignore_ascii_case(target_column.data_type.trim())
{
return Err(format!(
"Native Inceptor column '{}' needs matching source and target types ({} vs {})",
source_column.name, source_column.data_type, target_column.data_type
));
}
}
}
// Truncate target if overwrite mode (only when not rebuilding the table).
// When drop_target_before_create is true, the target table was just created
// and is already empty, so TRUNCATE is unnecessary.
@@ -9785,6 +9909,7 @@ where
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
) {
log::warn!("[transfer] upsert not supported for {:?}, falling back to append", target_db_type);
(TransferMode::Append, vec![])
@@ -9813,6 +9938,37 @@ where
let mut offset: u64 = 0;
let mut total_transferred: u64 = 0;
if server_side_complex_copy {
if is_cancelled(&request.transfer_id).await {
return Err("Cancelled".to_string());
}
let sql = transwarp_server_side_copy_sql(
&col_names,
&write_col_names,
table,
&request.source_schema,
&target_table,
&request.target_schema,
request.quote_target_column_names,
);
execute_on_pool(state, target_pool_key, &sql)
.await
.map_err(|error| format!("Native Inceptor transfer failed for '{table}': {error}"))?;
let copied = total_rows.expect("complex transfer count validated");
progress_callback(TransferProgress {
transfer_id: request.transfer_id.clone(),
table: table.to_string(),
table_index,
total_tables,
rows_transferred: copied,
total_rows,
status: TransferStatus::Running,
error: None,
terminal: false,
});
return Ok(copied);
}
// COPY fast path: PG-family append/overwrite transfers stream the whole
// table through the COPY protocol instead of paged SELECT + multi-row
// INSERT — no per-batch statement parsing, no JSON round-trip. Any failure
@@ -9895,9 +10051,11 @@ where
// does not hold up mid-table.
let mut keyset_indexes = transfer_keyset_column_indexes(&writable_columns, &primary_key_columns, source_db_type);
let mut keyset_cursor: Vec<serde_json::Value> = Vec::new();
// A single Agent cursor keeps Kyuubi/Impala rows in one query execution.
// A single Agent cursor keeps Hive-family rows in one query execution. Inceptor
// rejects the generic LIMIT/OFFSET form, just like the other Agent cursor paths.
// Re-running LIMIT/OFFSET pages is unstable for tables without a unique key.
let use_hive_server_cursor = matches!(source_db_type, DatabaseType::Kyuubi | DatabaseType::Impala);
let use_hive_server_cursor =
matches!(source_db_type, DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Transwarp);
let hive_server_transfer_sql = use_hive_server_cursor.then(|| {
transfer_cursor_sql(
&col_names,
@@ -17295,6 +17453,73 @@ SELECT 1 FROM dual"#
assert_eq!(sql, "INSERT INTO \"public\".\"users\" (\"id\", \"name\") VALUES\n(42, 'Ada')");
}
#[test]
fn inceptor_insert_uses_select_for_ordinary_tables() {
assert_eq!(SqlBatchLimits::for_database(&DatabaseType::Transwarp, 500).max_rows, 100);
let sql = generate_insert_typed(
&[String::from("id"), String::from("label")],
&[Some(String::from("int")), Some(String::from("string"))],
&[vec![json!(1), json!("first")], vec![json!(2), json!("second")]],
"events",
"analytics",
&DatabaseType::Transwarp,
None,
);
assert_eq!(
sql,
"INSERT INTO `analytics`.`events` (`id`, `label`)\nSELECT 1, 'first'\nUNION ALL\nSELECT 2, 'second'"
);
let batches = generate_insert_typed_sql_batches(
&[String::from("id"), String::from("label")],
&[Some(String::from("int")), Some(String::from("string"))],
&[vec![json!(1), json!("first")], vec![json!(2), json!("second")]],
"events",
"analytics",
&DatabaseType::Transwarp,
None,
SqlBatchLimits { max_rows: 100, target_sql_bytes: 76, hard_sql_bytes: Some(76) },
)
.unwrap();
assert_eq!(batches.iter().map(|(_, rows)| *rows).sum::<usize>(), 2);
assert!(batches.iter().all(|(sql, _)| sql.len() <= 76 && sql.contains("\nSELECT ")));
}
#[test]
fn inceptor_native_complex_copy_uses_server_side_projection() {
assert!(is_transwarp_complex_type("ARRAY<INT>"));
assert!(is_transwarp_complex_type("map<string,int>"));
assert!(is_transwarp_complex_type("struct<a:int>"));
assert!(!is_transwarp_complex_type("STRING"));
let columns = vec![db::ColumnInfo { data_type: "ARRAY<INT>".into(), ..Default::default() }];
assert!(!should_use_transwarp_server_side_copy(
&DatabaseType::Transwarp,
&TransferContent::StructureOnly,
&columns,
));
assert!(should_use_transwarp_server_side_copy(
&DatabaseType::Transwarp,
&TransferContent::StructureAndData,
&columns,
));
assert!(should_use_transwarp_server_side_copy(&DatabaseType::Transwarp, &TransferContent::DataOnly, &columns,));
assert!(!should_use_transwarp_server_side_copy(
&DatabaseType::Hive,
&TransferContent::StructureAndData,
&columns,
));
let sql = transwarp_server_side_copy_sql(
&["id".into(), "numbers".into(), "tags".into()],
&["id".into(), "numbers".into(), "tags".into()],
"events",
"source_db",
"events",
"target_db",
true,
);
assert_eq!(sql, "INSERT INTO `target_db`.`events` (`id`, `numbers`, `tags`) SELECT `id`, `numbers`, `tags` FROM `source_db`.`events`");
}
#[test]
fn postgres_insert_can_skip_duplicate_rows() {
let batches = generate_insert_typed_sql_batches_from_value_rows_with_options(
+1 -1
View File
@@ -2899,7 +2899,7 @@ async fn execute_sql_statement_with_options_typed_inner(
|| crate::sql::has_executable_sql(sql),
|db_type| crate::sql::has_executable_sql_for_database(sql, db_type),
);
if !has_executable_sql {
if !has_executable_sql && options.result_session_id.is_none() {
return Ok(empty_query_result(0));
}
@@ -1520,6 +1520,7 @@ mod tests {
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -52,6 +52,7 @@ fn change_options(table_name: &str, columns: Vec<EditableStructureColumn>) -> Ta
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -48,6 +48,7 @@ fn change_options(table_name: &str, columns: Vec<EditableStructureColumn>) -> Ta
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -62,6 +62,7 @@ async fn live_postgres_concurrent_index_builds_valid_index() {
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -195,6 +196,7 @@ async fn live_postgres_partitioned_parent_concurrent_request_rejected() {
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: true,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -314,6 +314,7 @@ async fn live_postgres_create_partitioned_table_executes() {
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -1017,6 +1017,7 @@ async fn live_sqlserver_table_structure_default_changes_drop_existing_constraint
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -73,6 +73,15 @@ pub fn label_for_key(agent_key: &str) -> Option<&'static str> {
mod tests {
use super::*;
#[test]
fn transwarp_products_share_one_driver_package() {
for profile in [None, Some("transwarp-inceptor")] {
assert_eq!(agent_key(&DatabaseType::Transwarp, profile), Some("transwarp"));
}
assert_eq!(driver_store_entries().filter(|(key, _)| *key == "transwarp").count(), 1);
assert!(!driver_store_entries().any(|(key, _)| key.starts_with("transwarp-")));
}
#[test]
fn h2_profiles_share_the_same_agent() {
assert_eq!(agent_key(&DatabaseType::H2, None), Some("h2"));
@@ -128,6 +128,14 @@ pub fn agent_connect_params_with_role(
sap_hana_jdbc_connection_string(config, host, port, database)
} else if matches!(config.db_type, DatabaseType::Trino | DatabaseType::PrestoSql) {
trino_like_jdbc_connection_string(config, host, port, database)?
} else if config.db_type == DatabaseType::Transwarp {
match config.connection_string.as_deref().filter(|value| !value.trim().is_empty()) {
Some(url) if host != config.host || port != config.port => {
crate::models::connection::rewrite_jdbc_url_host(url, host, port)?
}
Some(url) => url.to_string(),
None => String::new(),
}
} else if config.db_type == DatabaseType::H2 {
h2_agent_jdbc_connection_string(config)
} else {
@@ -1409,6 +1417,22 @@ mod tests {
assert_eq!(params["jdbc_driver_class"], "com.example.CustomDriver");
}
#[test]
fn transwarp_jdbc_url_follows_resolved_tunnel_endpoint() {
let mut cfg = config(DatabaseType::Transwarp, Some("analytics"));
cfg.host = "quark.example.com".to_string();
cfg.port = 10000;
cfg.connection_string = Some("jdbc:inceptor2://quark.example.com:10000/analytics;fetchSize=500".to_string());
cfg.jdbc_driver_class = Some("org.apache.hive.jdbc.HiveDriver".to_string());
cfg.jdbc_driver_paths = vec!["C:/drivers/inceptor-sdk.jar".to_string()];
let params = agent_connect_params(&cfg, "127.0.0.1", 18431, "analytics").unwrap();
assert_eq!(params["connection_string"], "jdbc:inceptor2://127.0.0.1:18431/analytics;fetchSize=500");
assert_eq!(params["jdbc_driver_class"], "org.apache.hive.jdbc.HiveDriver");
assert_eq!(params["jdbc_driver_paths"], serde_json::json!(["C:/drivers/inceptor-sdk.jar"]));
}
#[test]
fn db2_agent_connect_params_ignore_driver_registry_ids() {
for registry_id in ["db2", "DB2", " db2 ", "db2_zos", "db2i"] {
+2
View File
@@ -278,6 +278,7 @@ impl SqlDialectProfile {
// Oracle for statement splitting so semicolons inside the procedure body are not
// misinterpreted as client-side statement terminators.
| DatabaseType::Argo
| DatabaseType::Transwarp
)
}
}
@@ -4097,6 +4098,7 @@ BEGIN
END;";
assert_eq!(split_sql_statements_for_database(sql, DatabaseType::Argo), vec![sql.to_string()]);
assert_eq!(split_sql_statements_for_database(sql, DatabaseType::Transwarp), vec![sql.to_string()]);
}
#[test]
+69 -2
View File
@@ -340,6 +340,8 @@ pub struct DataGridConditionalUpdateSqlOptions {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HiveTablePropertiesSqlOptions {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database_type: Option<DatabaseType>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table_name: String,
@@ -1159,6 +1161,15 @@ pub fn build_data_grid_conditional_update_sql(options: DataGridConditionalUpdate
}
pub fn build_hive_table_properties_sql(options: HiveTablePropertiesSqlOptions) -> String {
if options.database_type == Some(DatabaseType::Transwarp)
&& options.property_name.eq_ignore_ascii_case("transactional")
{
let database = options.schema.as_deref().unwrap_or("default").replace('\'', "''");
let table = options.table_name.replace('\'', "''");
return format!(
"SELECT transactional FROM system.tables_v WHERE database_name = '{database}' AND table_name = '{table}'"
);
}
let table = qualified_table_name(Some(DatabaseType::Hive), options.schema.as_deref(), &options.table_name);
let property = options.property_name.replace('\'', "''");
format!("SHOW TBLPROPERTIES {table} ('{property}')")
@@ -1755,7 +1766,11 @@ fn build_data_grid_save_statements(
.join(", ");
statements.push(data_grid_statement(
options.database_type,
format!("INSERT INTO {table} ({columns}) VALUES ({values})"),
if options.database_type == Some(DatabaseType::Transwarp) {
format!("INSERT INTO {table} ({columns}) SELECT {values}")
} else {
format!("INSERT INTO {table} ({columns}) VALUES ({values})")
},
));
}
@@ -1875,7 +1890,11 @@ fn build_data_grid_rollback_statements(
} else {
statements.push(data_grid_statement(
options.database_type,
format!("INSERT INTO {table} ({columns}) VALUES ({values})"),
if options.database_type == Some(DatabaseType::Transwarp) {
format!("INSERT INTO {table} ({columns}) SELECT {values}")
} else {
format!("INSERT INTO {table} ({columns}) VALUES ({values})")
},
));
}
}
@@ -6435,6 +6454,7 @@ mod tests {
fn builds_hive_table_properties_sql() {
assert_eq!(
build_hive_table_properties_sql(HiveTablePropertiesSqlOptions {
database_type: None,
schema: Some("default".to_string()),
table_name: "events".to_string(),
property_name: "transactional".to_string(),
@@ -6572,6 +6592,53 @@ mod tests {
assert_eq!(effective_columns(&postgres_options), vec![Some("events.id".to_string())]);
}
#[test]
fn transwarp_grid_insert_and_restore_use_select_syntax() {
let options = DataGridSaveStatementOptions {
database_type: Some(DatabaseType::Transwarp),
identifier_quote: None,
server_version: None,
table_meta: DataGridTableMeta {
catalog: None,
database: None,
schema: Some("default".to_string()),
table_name: "events".to_string(),
primary_keys: vec![],
columns: Some(vec![column("id", "int", true, None), column("name", "string", true, None)]),
},
columns: vec!["id".to_string(), "name".to_string()],
source_columns: None,
rows: vec![vec![json!(1), json!("before")]],
dirty_rows: vec![],
deleted_rows: vec![0],
new_rows: vec![vec![json!(2), json!("after")]],
include_database_name: false,
};
let result = prepare_data_grid_save(options);
assert!(result
.statements
.iter()
.any(|sql| sql.contains("INSERT INTO `default`.`events` (`id`, `name`) SELECT 2, 'after'")));
assert!(result
.rollback_statements
.iter()
.any(|sql| sql.contains("INSERT INTO `default`.`events` (`id`, `name`) SELECT 1, 'before'")));
}
#[test]
fn transwarp_transactional_property_uses_system_table() {
let sql = build_hive_table_properties_sql(HiveTablePropertiesSqlOptions {
database_type: Some(DatabaseType::Transwarp),
schema: Some("analytics".to_string()),
table_name: "events".to_string(),
property_name: "transactional".to_string(),
});
assert_eq!(
sql,
"SELECT transactional FROM system.tables_v WHERE database_name = 'analytics' AND table_name = 'events'"
);
}
#[test]
fn formats_temporal_copy_literals() {
assert_eq!(
@@ -56,6 +56,7 @@ pub fn is_schema_aware(database_type: DatabaseType) -> bool {
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
@@ -894,8 +894,8 @@ pub fn profile_for(db_type: DatabaseType) -> DdlDialectProfile {
profile
}
DuckDb | Questdb | Snowflake | Trino | PrestoSql | Hive | Kyuubi | Impala | Argo | Spark | Bigquery
| Spanner | Kylin | Ignite | Ignite3 | Oscar | Tdengine | Iotdb | Databricks | Jdbc => {
DuckDb | Questdb | Snowflake | Trino | PrestoSql | Hive | Kyuubi | Impala | Argo | Transwarp | Spark
| Bigquery | Spanner | Kylin | Ignite | Ignite3 | Oscar | Tdengine | Iotdb | Databricks | Jdbc => {
conservative_ansi(db_type)
}
@@ -152,6 +152,7 @@ pub fn quote_table_identifier(database_type: Option<DatabaseType>, name: &str) -
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
| DatabaseType::Spark
| DatabaseType::Databricks
| DatabaseType::Databend
@@ -428,6 +429,7 @@ pub fn quote_transfer_identifier(name: &str, database_type: &DatabaseType) -> St
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
| DatabaseType::Spark
| DatabaseType::Questdb => format!("`{}`", name.replace('`', "``")),
DatabaseType::SqlServer => format!("[{}]", name.replace(']', "]]")),
@@ -782,7 +782,13 @@ pub(super) fn build_select_columns(
// natively and users can narrow the projection by editing the SQL.
if !matches!(
database_type,
Some(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo)
Some(
DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Transwarp
)
) {
return "*".to_string();
}
+68 -1
View File
@@ -279,6 +279,12 @@ fn build_create_database_statement(options: &CreateDatabaseSqlOptions) -> Result
if !supports_create_database_target(options.database_type, options.driver_profile.as_deref()) {
return Err(format!("Creating databases is not supported for {}.", database_label(options.database_type)));
}
if options.database_type == Some(DatabaseType::Transwarp) {
return Ok(format!(
"CREATE DATABASE IF NOT EXISTS {};",
quote_table_identifier(options.database_type, &options.name)
));
}
if is_informix_family(options.database_type, options.driver_profile.as_deref()) {
// Informix / GBase 8s accept only a bare `CREATE DATABASE <name>`. The new database
// inherits the instance default locale, and the MySQL `CHARACTER SET`/`COLLATE`
@@ -347,6 +353,7 @@ pub fn supports_create_database_target(database_type: Option<DatabaseType>, driv
| DatabaseType::Highgo
| DatabaseType::Kingbase
| DatabaseType::Yashandb
| DatabaseType::Transwarp
)
)
}
@@ -691,7 +698,11 @@ fn supports_truncate_table_cascade(database_type: Option<DatabaseType>) -> bool
}
pub fn build_drop_database_sql(options: DatabaseNameSqlOptions) -> String {
format!("DROP DATABASE {};", quote_table_identifier(options.database_type, &options.name))
let name = quote_table_identifier(options.database_type, &options.name);
if options.database_type == Some(DatabaseType::Transwarp) {
return format!("DROP DATABASE IF EXISTS {name};");
}
format!("DROP DATABASE {name};")
}
pub fn build_update_database_properties_sql(options: DatabasePropertyEditSqlOptions) -> Result<String, String> {
@@ -921,6 +932,9 @@ pub fn supports_object_rename(database_type: Option<DatabaseType>, object_type:
if database_type == DatabaseType::SqlServer {
return true;
}
if database_type == DatabaseType::Transwarp {
return object_type == DatabaseObjectType::Table;
}
if matches!(object_type, DatabaseObjectType::Procedure | DatabaseObjectType::Function) {
return false;
}
@@ -964,6 +978,14 @@ pub fn build_rename_object_sql(options: RenameObjectSqlOptions) -> Result<String
));
}
if database_type == Some(DatabaseType::Transwarp) {
return Ok(format!(
"ALTER TABLE {} RENAME TO {};",
qualified_name(database_type, options.schema.as_deref(), &options.old_name),
qualified_name(database_type, options.schema.as_deref(), &options.new_name)
));
}
if matches!(database_type, Some(DatabaseType::Mysql | DatabaseType::Goldendb)) {
return Ok(format!(
"RENAME TABLE {} TO {};",
@@ -1539,6 +1561,50 @@ mod tests {
assert!(!supports_create_database_target(Some(DatabaseType::Gbase), Some("gbase8a")));
assert!(!supports_create_database_target(Some(DatabaseType::Gbase), None));
assert!(supports_create_database_target(Some(DatabaseType::Informix), None));
assert!(supports_create_database_target(Some(DatabaseType::Transwarp), Some("transwarp-inceptor")));
}
#[test]
fn transwarp_database_actions_match_waterdrop_ddl() {
assert!(!supports_create_schema_target(Some(DatabaseType::Transwarp)));
assert_eq!(
build_create_database_sql(CreateDatabaseSqlOptions {
database_type: Some(DatabaseType::Transwarp),
driver_profile: Some("transwarp-inceptor".to_string()),
target: None,
parent: None,
name: "analytics db".to_string(),
charset: None,
collation: None,
})
.unwrap(),
"CREATE DATABASE IF NOT EXISTS `analytics db`;"
);
assert_eq!(
build_drop_database_sql(DatabaseNameSqlOptions {
database_type: Some(DatabaseType::Transwarp),
name: "analytics db".to_string(),
}),
"DROP DATABASE IF EXISTS `analytics db`;"
);
}
#[test]
fn transwarp_table_rename_matches_waterdrop_ddl() {
assert!(supports_object_rename(Some(DatabaseType::Transwarp), DatabaseObjectType::Table));
assert!(!supports_object_rename(Some(DatabaseType::Transwarp), DatabaseObjectType::View));
assert!(!supports_database_rename(Some(DatabaseType::Transwarp)));
assert_eq!(
build_rename_object_sql(RenameObjectSqlOptions {
database_type: Some(DatabaseType::Transwarp),
object_type: DatabaseObjectType::Table,
schema: Some("analytics".to_string()),
old_name: "old table".to_string(),
new_name: "new table".to_string(),
})
.unwrap(),
"ALTER TABLE `analytics`.`old table` RENAME TO `analytics`.`new table`;"
);
}
#[test]
@@ -2938,6 +3004,7 @@ mod tests {
table_comment: None,
original_table_comment: None,
mysql_engine: None,
transwarp_create: None,
partitioned: false,
is_gaussdb_m_mode: false,
table_collation: None,
@@ -9,6 +9,7 @@ mod indexes;
mod mysql_engine;
mod owner;
mod partitions;
mod transwarp;
mod triggers;
mod types;
mod util;
@@ -372,6 +372,15 @@ pub(super) fn build_mysql_existing_column_clause(column: &EditableStructureColum
format!("{operation}{position_clause}")
}
pub(super) fn build_transwarp_existing_column_clause(column: &EditableStructureColumn) -> String {
let original_name = column.original.as_ref().map(|original| original.name.as_str()).unwrap_or(&column.name);
format!(
"CHANGE {} {}",
quote_ident(StructureDialect::Mysql, original_name),
column_definition(StructureDialect::Mysql, column)
)
}
pub(super) fn build_doris_existing_column_sql(
table: &str,
column: &EditableStructureColumn,
@@ -1,3 +1,4 @@
use super::column_alter::build_transwarp_existing_column_clause;
use super::column_alter::{
build_clickhouse_existing_column_sql, build_dameng_existing_column_sql, build_doris_existing_column_sql,
build_duckdb_existing_column_sql, build_h2_existing_column_sql, build_informix_existing_column_sql,
@@ -115,7 +116,8 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
}
let active_index = active_columns.iter().position(|active| active.id == column.id).unwrap_or(0);
let position_clause = if has_original_column_positions {
let position_clause = if has_original_column_positions && options.database_type != Some(DatabaseType::Transwarp)
{
column_position_clause(dialect, &active_columns, active_index)
} else {
String::new()
@@ -248,10 +250,12 @@ pub(super) fn build_column_sql(options: &TableStructureSqlOptions, warnings: &mu
} else {
column
};
let clause = build_mysql_existing_column_clause(
effective_column,
if has_position_change { &position_clause } else { "" },
);
let position = if has_position_change { &position_clause } else { "" };
let clause = if options.database_type == Some(DatabaseType::Transwarp) {
build_transwarp_existing_column_clause(effective_column)
} else {
build_mysql_existing_column_clause(effective_column, position)
};
if mysql_coalesced_primary_key_change
.is_some_and(|change| mysql_auto_increment_touches_primary_key(column, change))
{
@@ -592,7 +596,9 @@ pub(super) fn build_add_column_sql(
driver_profile: Option<&str>,
) -> Vec<String> {
let definition = column_definition(dialect, column);
let mut statements = if is_oracle_like(dialect) || dialect == StructureDialect::Informix {
let mut statements = if database_type == Some(DatabaseType::Transwarp) {
vec![format!("ALTER TABLE {table} ADD COLUMNS ({definition});")]
} else if is_oracle_like(dialect) || dialect == StructureDialect::Informix {
vec![format!("ALTER TABLE {table} ADD ({definition});")]
} else {
let add_keyword = if dialect == StructureDialect::SqlServer
@@ -1,6 +1,7 @@
use super::dialect::{capabilities_for, database_label, dialect_label, StructureDialect};
use super::types::TableStructureSqlOptions;
use super::util::{clean, qualified_table, quote_string};
use crate::models::connection::DatabaseType;
pub(super) fn build_table_comment_sql(options: &TableStructureSqlOptions, warnings: &mut Vec<String>) -> Vec<String> {
let capabilities = capabilities_for(options.database_type, options.driver_profile.as_deref());
@@ -19,6 +20,9 @@ pub(super) fn build_table_comment_sql(options: &TableStructureSqlOptions, warnin
let dialect = capabilities.dialect;
let table = qualified_table(dialect, options.schema.as_deref(), &options.table_name);
let quoted = quote_string(&clean(new_comment));
if options.database_type == Some(DatabaseType::Transwarp) {
return vec![format!("ALTER TABLE {table} SET TBLPROPERTIES ('comment' = {quoted});")];
}
match dialect {
StructureDialect::Mysql | StructureDialect::GaussdbM => {
vec![format!("ALTER TABLE {table} COMMENT = {quoted};")]
@@ -8,6 +8,7 @@ use super::dialect::{capabilities_for, database_label, StructureDialect};
use super::foreign_keys::build_foreign_key_sql_for_new_table;
use super::indexes::build_create_index_statements;
use super::mysql_engine::{append_mysql_table_option, validate_mysql_engine};
use super::transwarp;
use super::triggers::build_trigger_sql_for_new_table;
use super::types::{TableStructureSqlOptions, TableStructureSqlResult};
use super::util::{
@@ -63,6 +64,7 @@ pub(super) fn build_create_table_sql_with_partition_clause(
}
validate_columns(&active_columns, &mut warnings);
validate_dameng_identity(&options, &active_columns, &mut warnings);
transwarp::validate_create_options(&options, &active_columns, &mut warnings);
if !warnings.is_empty() {
return TableStructureSqlResult { statements: Vec::new(), warnings };
}
@@ -105,6 +107,9 @@ pub(super) fn build_create_table_sql_with_partition_clause(
let mut column_definitions = Vec::new();
for column in &active_columns {
if transwarp::is_partition_column(&options, column) {
continue;
}
let mut data_type = column_data_type(dialect, column);
// SQLite accepts AUTOINCREMENT only on an exact INTEGER PRIMARY KEY,
// so integer-family aliases are normalized when auto-increment is on.
@@ -190,7 +195,8 @@ pub(super) fn build_create_table_sql_with_partition_clause(
column_definitions.push(format!("PRIMARY KEY ({pk_list})"));
}
let create_table = format!("CREATE TABLE {table} (\n {}\n)", column_definitions.join(",\n "));
let mut create_table = format!("CREATE TABLE {table} (\n {}\n)", column_definitions.join(",\n "));
create_table.push_str(&transwarp::create_table_suffix(&options, &active_columns, dialect));
let create_table = match partition_clause {
Some(clause) => format!("{create_table} {clause}"),
None => create_table,
@@ -205,7 +211,7 @@ pub(super) fn build_create_table_sql_with_partition_clause(
if capabilities.comment {
let table_comment = clean(options.table_comment.as_deref().unwrap_or(""));
if !table_comment.is_empty() {
if !table_comment.is_empty() && options.database_type != Some(DatabaseType::Transwarp) {
if matches!(dialect, StructureDialect::Mysql | StructureDialect::GaussdbM) {
if let Some(last) = statements.last_mut() {
append_mysql_table_option(last, &format!("COMMENT = {}", quote_string(&table_comment)));
@@ -379,6 +379,15 @@ pub(super) fn capabilities_for(
rebuild_index: true,
..base
},
// Inceptor accepts ADD COLUMNS and CHANGE, but not DROP COLUMN.
Some(DatabaseType::Transwarp) => TableStructureCapabilities {
dialect: StructureDialect::Mysql,
add_column: true,
rename_column: true,
alter_existing_column: true,
comment: true,
..base
},
_ => base,
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,124 @@
use super::column_format::column_data_type;
use super::dialect::StructureDialect;
use super::types::{EditableStructureColumn, TableStructureSqlOptions, TranswarpCreateTableOptions};
use super::util::{clean, quote_new_ident, quote_string};
use crate::models::connection::DatabaseType;
pub(super) fn validate_create_options(
options: &TableStructureSqlOptions,
active_columns: &[&EditableStructureColumn],
warnings: &mut Vec<String>,
) {
let Some(create) = options.transwarp_create.as_ref() else { return };
if options.database_type != Some(DatabaseType::Transwarp) {
warnings.push("Inceptor table options require a Transwarp connection.".to_string());
return;
}
for name in create.partition_columns.iter().chain(&create.bucket_columns) {
if !active_columns.iter().any(|column| column.name.eq_ignore_ascii_case(name.trim())) {
warnings.push(format!("Inceptor column {name:?} does not exist in the table draft."));
}
}
if create.partition_columns.len() >= active_columns.len() && !create.partition_columns.is_empty() {
warnings.push("Inceptor partitioned tables need at least one non-partition column.".to_string());
}
for names in [&create.partition_columns, &create.bucket_columns] {
let mut seen = std::collections::HashSet::new();
for name in names {
if name.trim().is_empty() || !seen.insert(name.trim().to_ascii_lowercase()) {
warnings.push("Inceptor partition and bucket columns must be distinct and non-empty.".to_string());
break;
}
}
}
if create
.bucket_columns
.iter()
.any(|bucket| create.partition_columns.iter().any(|partition| partition.eq_ignore_ascii_case(bucket)))
{
warnings.push("Inceptor bucket columns cannot also be partition columns.".to_string());
}
if create.bucket_columns.is_empty() != create.bucket_count.is_none()
|| create.bucket_count.is_some_and(|count| count == 0 || count > 4096)
{
warnings.push("Inceptor bucketing requires columns and a bucket count from 1 to 4096.".to_string());
}
let storage = create.storage_format.as_deref().unwrap_or("").trim();
if !storage.is_empty()
&& !["ORC", "PARQUET", "TEXTFILE"].iter().any(|allowed| storage.eq_ignore_ascii_case(allowed))
{
warnings.push("Unsupported Inceptor storage format.".to_string());
}
if create.transactional && (!storage.eq_ignore_ascii_case("ORC") || create.bucket_columns.is_empty()) {
warnings.push("Inceptor transactional tables require ORC storage and bucket columns.".to_string());
}
}
pub(super) fn is_partition_column(options: &TableStructureSqlOptions, column: &EditableStructureColumn) -> bool {
options.database_type == Some(DatabaseType::Transwarp)
&& options.transwarp_create.as_ref().is_some_and(|create| {
create.partition_columns.iter().any(|name| column.name.eq_ignore_ascii_case(name.trim()))
})
}
pub(super) fn create_table_suffix(
options: &TableStructureSqlOptions,
active_columns: &[&EditableStructureColumn],
dialect: StructureDialect,
) -> String {
if options.database_type != Some(DatabaseType::Transwarp) {
return String::new();
}
let mut suffix = String::new();
let comment = clean(options.table_comment.as_deref().unwrap_or(""));
if !comment.is_empty() {
suffix.push_str(&format!(" COMMENT {}", quote_string(&comment)));
}
if let Some(create) = options.transwarp_create.as_ref() {
suffix.push_str(&create_options_clause(create, active_columns, dialect));
}
suffix
}
fn create_options_clause(
create: &TranswarpCreateTableOptions,
active_columns: &[&EditableStructureColumn],
dialect: StructureDialect,
) -> String {
let mut clauses = String::new();
if !create.partition_columns.is_empty() {
let partition_columns = create
.partition_columns
.iter()
.map(|name| {
let column = active_columns
.iter()
.find(|column| column.name.eq_ignore_ascii_case(name.trim()))
.expect("validated partition column");
format!(
"{} {}",
quote_new_ident(Some(DatabaseType::Transwarp), dialect, &column.name),
column_data_type(dialect, column)
)
})
.collect::<Vec<_>>()
.join(", ");
clauses.push_str(&format!(" PARTITIONED BY ({partition_columns})"));
}
if let Some(count) = create.bucket_count {
let columns = create
.bucket_columns
.iter()
.map(|name| quote_new_ident(Some(DatabaseType::Transwarp), dialect, name.trim()))
.collect::<Vec<_>>()
.join(", ");
clauses.push_str(&format!(" CLUSTERED BY ({columns}) INTO {count} BUCKETS"));
}
if let Some(storage) = create.storage_format.as_deref().map(str::trim).filter(|value| !value.is_empty()) {
clauses.push_str(&format!(" STORED AS {}", storage.to_ascii_uppercase()));
}
if create.transactional {
clauses.push_str(" TBLPROPERTIES ('transactional'='true')");
}
clauses
}
@@ -237,6 +237,8 @@ pub struct TableStructureSqlOptions {
pub original_table_comment: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mysql_engine: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub transwarp_create: Option<TranswarpCreateTableOptions>,
/// MySQL only: the table's current default collation
/// (`information_schema.TABLES.TABLE_COLLATION`). A column whose collation
/// merely matches it inherits the table default, so its `CHARACTER SET` /
@@ -259,6 +261,21 @@ pub struct TableStructureSqlOptions {
pub is_gaussdb_m_mode: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TranswarpCreateTableOptions {
#[serde(default)]
pub partition_columns: Vec<String>,
#[serde(default)]
pub bucket_columns: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bucket_count: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub storage_format: Option<String>,
#[serde(default)]
pub transactional: bool,
}
/// Options for `build_table_partition_operation_sql`.
///
/// Partition maintenance is deliberately kept out of `TableStructureSqlOptions`:
+2 -1
View File
@@ -1178,6 +1178,7 @@ impl ConnectionConfig {
DatabaseType::Trino => format!("trino://{host}:{port}{db_part}"),
DatabaseType::PrestoSql => format!("prestosql://{host}:{port}{db_part}"),
DatabaseType::Hive | DatabaseType::Argo => format!("hive://{host}:{port}{db_part}"),
DatabaseType::Transwarp => format!("transwarp://{host}:{port}{db_part}"),
DatabaseType::Kyuubi => format!("kyuubi://{host}:{port}{db_part}"),
DatabaseType::Impala => format!("impala://{host}:{port}{db_part}"),
DatabaseType::Spark => format!("spark://{host}:{port}{db_part}"),
@@ -1410,7 +1411,7 @@ impl ConnectionConfig {
DatabaseType::PrestoSql => {
format!("prestosql://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Hive | DatabaseType::Argo => {
DatabaseType::Hive | DatabaseType::Argo | DatabaseType::Transwarp => {
format!("hive://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Kyuubi => {
Binary file not shown.

After

Width:  |  Height:  |  Size: 53 KiB

+7
View File
@@ -22,3 +22,10 @@ test("Apache Phoenix database icon keeps its official source attribution", () =>
assert.match(svg, /#f7931d/i);
assert.match(svg, /#ed1c24/i);
});
test("Inceptor connection icon matches the Waterdrop provider asset", () => {
const png = readFileSync(path.resolve("apps/desktop/public/icons/database/transwarp-inceptor.png"));
assert.equal(png.subarray(1, 4).toString("ascii"), "PNG");
assert.equal(png.readUInt32BE(16), 64);
assert.equal(png.readUInt32BE(20), 64);
});
@@ -477,6 +477,14 @@ profiles:
user: ""
urlParams: auth=NONE
category: analytics
- id: transwarp-inceptor
dbType: transwarp
label: 星环Inceptor
icon: transwarp-inceptor
port: 10000
user: ""
category: domestic
pickerLabel: 星环Inceptor
- id: argo
dbType: argo
label: ArgoDB (Transwarp)
+41
View File
@@ -0,0 +1,41 @@
schemaVersion: 1
order: 516
dbType: transwarp
rustVariant: Transwarp
label: 星环Inceptor
dialect: Transwarp
runtimeMode: agent
mcpMode: bridge
agentKey: transwarp
driverStoreVisible: true
driverStoreOrder: 52
driverProfiles:
- profile: transwarp-inceptor
agentKey: transwarp
label: 星环Inceptor
storeVisible: false
singleConnectionPool: false
metadataConnectionScoped: false
skipTcpProbe: true
defaultPort: 10000
traits:
schemaAware: true
supportLevel: operate
capabilities:
queryExecution: true
metadataBrowse: true
objectBrowser: true
objectSource: true
schemaSearch: true
diagram: false
tableDataEdit: true
tableStructureEdit: true
tableImport: true
dataTransfer: true
sqlFileExecution: true
databaseCreate: true
fieldLineage: false
sqlExplain: false
userAdmin: false
driverManagement: true

Some files were not shown because too many files have changed in this diff Show More