feat(oceanbase): retain original CLOB snapshots for bounded value reads

This commit is contained in:
efeire
2026-10-09 18:04:45 +08:00
parent 65c4d16dca
commit 989dfb79bb
30 changed files with 1232 additions and 42 deletions
@@ -421,7 +421,7 @@ public abstract class AbstractJdbcAgent extends BaseDatabaseAgent {
releasePooledConnection(true);
return;
}
if (sessionAffinity || executor.hasOpenSessions() || executor.hasActiveStatements()) {
if (sessionAffinity || executor.hasOpenSessions() || executor.hasActiveStatements() || hasRetainedResultResources()) {
return;
}
releasePooledConnection(!preparePooledConnectionForReturn());
@@ -438,7 +438,7 @@ public abstract class AbstractJdbcAgent extends BaseDatabaseAgent {
if (sessionAffinity) {
return;
}
if (executor.hasOpenSessions() || executor.hasActiveStatements()) {
if (executor.hasOpenSessions() || executor.hasActiveStatements() || hasRetainedResultResources()) {
return;
}
releasePooledConnection(!preparePooledConnectionForReturn());
@@ -713,7 +713,12 @@ public abstract class AbstractJdbcAgent extends BaseDatabaseAgent {
);
}
protected boolean hasRetainedResultResources() { return false; }
protected void releaseRetainedResultResources() { }
private void closeCurrentConnection() throws Exception {
releaseRetainedResultResources();
if (pooledLease != null) {
boolean evict = pooledConnectionPoisoned || sessionAffinity || !preparePooledConnectionForReturn();
releasePooledConnection(evict);
@@ -733,6 +738,7 @@ public abstract class AbstractJdbcAgent extends BaseDatabaseAgent {
}
private void releasePooledConnection(boolean evict) {
releaseRetainedResultResources();
JdbcConnectionPoolRegistry.Lease lease = pooledLease;
pooledLease = null;
connection = null;
@@ -38,6 +38,8 @@ public final class AgentProtocol {
public static final String METHOD_EXECUTE_QUERY_PAGE = "execute_query_page";
public static final String METHOD_FETCH_QUERY_PAGE = "fetch_query_page";
public static final String METHOD_CLOSE_QUERY_SESSION = "close_query_session";
public static final String METHOD_READ_LARGE_VALUE_CHUNK = "read_large_value_chunk";
public static final String METHOD_RELEASE_LARGE_VALUE = "release_large_value";
public static final String METHOD_START_TABLE_READ = "start_table_read";
public static final String METHOD_FETCH_TABLE_READ_PAGE = "fetch_table_read_page";
public static final String METHOD_CLOSE_TABLE_READ_SESSION = "close_table_read_session";
@@ -215,6 +217,8 @@ public final class AgentProtocol {
METHOD_EXECUTE_QUERY_PAGE,
METHOD_FETCH_QUERY_PAGE,
METHOD_CLOSE_QUERY_SESSION,
METHOD_READ_LARGE_VALUE_CHUNK,
METHOD_RELEASE_LARGE_VALUE,
METHOD_START_TABLE_READ,
METHOD_FETCH_TABLE_READ_PAGE,
METHOD_CLOSE_TABLE_READ_SESSION,
@@ -204,6 +204,14 @@ public interface DatabaseAgent {
return AgentExecutionContext.jdbcExecutor().closeQuerySession(sessionId);
}
default Object readLargeValueChunk(String valueRef, long offset, int limit) {
throw new UnsupportedOperationException("LOB snapshot chunks are not supported by this agent");
}
default boolean releaseLargeValue(String valueRef) { return false; }
default void invalidateLargeValues() { }
default QueryPageResult startTableRead(String sql, String schema, QueryPageOptions options) {
Connection conn = getConnection();
if (conn == null) {
@@ -1143,6 +1143,20 @@ public final class JdbcExecutor {
* agents to convert vendor-specific column types (e.g. PostGIS
* {@code geometry}) without re-querying the metadata.
*/
public <T> T withActiveStatement(Statement statement, SqlOperation<T> operation) throws SQLException {
activeStatements.add(statement);
try {
return operation.run();
} finally {
activeStatements.remove(statement);
}
}
@FunctionalInterface
public interface SqlOperation<T> {
T run() throws SQLException;
}
public interface ColumnAwareResultValueReader extends ResultValueReader {
Object read(ResultSet rs, int index, int sqlType, String columnTypeName) throws SQLException;
@@ -152,6 +152,7 @@ public final class JsonRpcServer {
void cancelActiveStatements() {
jdbcExecutor.cancelActiveStatements();
agent.invalidateLargeValues();
AbstractJdbcAgent jdbcAgent = pooledJdbcAgent();
if (jdbcAgent != null) {
jdbcAgent.releaseIdlePooledConnection(jdbcExecutor);
@@ -321,6 +322,13 @@ public final class JsonRpcServer {
if (AgentProtocol.METHOD_CLOSE_QUERY_SESSION.equals(method)) {
return agent.closeQuerySession(params.get("sessionId").getAsString());
}
if (AgentProtocol.METHOD_READ_LARGE_VALUE_CHUNK.equals(method)) {
return agent.readLargeValueChunk(params.get("valueRef").getAsString(),
params.get("offset").getAsLong(), intOrDefault(params, "limit", 4096));
}
if (AgentProtocol.METHOD_RELEASE_LARGE_VALUE.equals(method)) {
return agent.releaseLargeValue(params.get("valueRef").getAsString());
}
if (AgentProtocol.METHOD_START_TABLE_READ.equals(method)) {
return agent.startTableRead(
params.get("sql").getAsString(),
@@ -357,20 +365,25 @@ public final class JsonRpcServer {
return result;
}
if (AgentProtocol.METHOD_EXECUTE_TRANSACTION.equals(method)) {
agent.invalidateLargeValues();
Type statementsType = new TypeToken<List<String>>() {}.getType();
List<String> statements = gson.fromJson(params.get("statements"), statementsType);
return agent.executeTransaction(statements, stringOrNull(params, "schema"));
}
if (AgentProtocol.METHOD_BEGIN_MANUAL_TRANSACTION.equals(method)) {
agent.invalidateLargeValues();
return agent.beginManualTransaction(stringOrNull(params, "schema"));
}
if (AgentProtocol.METHOD_COMMIT_MANUAL_TRANSACTION.equals(method)) {
agent.invalidateLargeValues();
return agent.commitManualTransaction();
}
if (AgentProtocol.METHOD_ROLLBACK_MANUAL_TRANSACTION.equals(method)) {
agent.invalidateLargeValues();
return agent.rollbackManualTransaction();
}
if (AgentProtocol.METHOD_EXECUTE_BATCH.equals(method)) {
agent.invalidateLargeValues();
Type statementsType = new TypeToken<List<String>>() {}.getType();
List<String> statements = gson.fromJson(params.get("statements"), statementsType);
return agent.executeBatch(statements, stringOrNull(params, "schema"));
@@ -0,0 +1,49 @@
package com.dbx.agent;
import com.google.gson.JsonParser;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
class JsonRpcLargeValueTest {
@Test
void dispatchesOpaqueLocatorAndCharacterOffsetAndReleasesItWithoutSql() throws Exception {
List<String> calls = new ArrayList<>();
DatabaseAgent agent = (DatabaseAgent) Proxy.newProxyInstance(DatabaseAgent.class.getClassLoader(),
new Class<?>[]{DatabaseAgent.class}, (object, method, args) -> {
if (method.getName().equals("supportsQueryTiming")) return false;
if (method.getName().equals("readLargeValueChunk")) {
calls.add("read");
assertArrayEquals(new Object[]{"opaque-original", 700L, 4096}, args);
return Map.of("status", "ok", "data", "中文😀", "next_offset", 703L, "eof", false, "value_kind", "text");
}
if (method.getName().equals("releaseLargeValue")) { calls.add("release"); assertEquals("opaque-original", args[0]); return true; }
return null;
});
JsonRpcServer server = new JsonRpcServer(agent);
Object result = server.dispatchForRuntime(AgentProtocol.METHOD_READ_LARGE_VALUE_CHUNK,
JsonParser.parseString("{\"valueRef\":\"opaque-original\",\"offset\":700,\"limit\":4096}").getAsJsonObject());
assertEquals("中文😀", ((Map<?, ?>) result).get("data"));
assertEquals(true, server.dispatchForRuntime(AgentProtocol.METHOD_RELEASE_LARGE_VALUE,
JsonParser.parseString("{\"valueRef\":\"opaque-original\"}").getAsJsonObject()));
assertEquals(List.of("read", "release"), calls);
}
@Test
void invalidatesLocatorsBeforeTransactionBoundary() throws Exception {
List<String> calls = new ArrayList<>();
DatabaseAgent agent = (DatabaseAgent) Proxy.newProxyInstance(DatabaseAgent.class.getClassLoader(),
new Class<?>[]{DatabaseAgent.class}, (object, method, args) -> {
if (method.getName().equals("supportsQueryTiming")) return false;
if (method.getName().equals("invalidateLargeValues")) { calls.add("invalidate"); return null; }
if (method.getName().equals("commitManualTransaction")) { calls.add("commit"); return Map.of("ok", true); }
return null;
});
new JsonRpcServer(agent).dispatchForRuntime(AgentProtocol.METHOD_COMMIT_MANUAL_TRANSACTION,
JsonParser.parseString("{}").getAsJsonObject());
assertEquals(List.of("invalidate", "commit"), calls);
}
}
@@ -0,0 +1,214 @@
package com.dbx.agent.oceanbaseoracle;
import com.dbx.agent.JdbcExecutor;
import com.oceanbase.jdbc.OceanBaseStatement;
import java.nio.ByteBuffer;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.sql.CallableStatement;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
/** Captured locators belong to this agent's pinned connection, never to a new SELECT. */
final class OceanBaseLobValues {
static final String MARKER_PREFIX = "__DBX_LARGE_VALUE_BYTES_";
static final int PREVIEW_CHARACTERS = 256;
static final int MAX_CHUNK_CHARACTERS = 4096;
private static final int MAX_REFS = 4096;
private static final long IDLE_MILLIS = 5 * 60_000L;
private static final long MAX_AGE_MILLIS = 30 * 60_000L;
private final Map<String, Entry> entries = new LinkedHashMap<>();
private final BlockReader reader;
@FunctionalInterface
interface BlockReader {
Chunk read(Connection connection, Clob locator, long offset, int limit) throws SQLException;
}
OceanBaseLobValues() { this(OceanBaseLobValues::read); }
OceanBaseLobValues(BlockReader reader) { this.reader = reader; }
record Preview(String text, String ref) {}
record Chunk(String status, String data, long next_offset, boolean eof, String value_kind) {}
record MarkedRows(List<String> columns, List<String> types, List<List<Object>> rows) {}
private static final class Entry {
final Connection connection;
final Clob locator;
final long created = System.currentTimeMillis();
long accessed = created;
Entry(Connection connection, Clob locator) {
this.connection = connection;
this.locator = locator;
}
}
synchronized Object preview(ResultSet rs, int index, int sqlType, String typeName) throws SQLException {
if (!isCharacterLob(sqlType, typeName) || hasMarkerCollision(rs)) return null;
// NCLOB's charset form must be verified separately; retain the complete existing read.
if (sqlType == Types.NCLOB || "NCLOB".equalsIgnoreCase(typeName)) return null;
Clob locator = rs.getClob(index);
if (locator == null || rs.wasNull()) return new Preview(null, null);
Connection connection = rs.getStatement().getConnection();
prune();
if (entries.size() >= MAX_REFS) {
locator.free();
throw new SQLException("LOB result resource limit reached; close older results and execute again", "HY001");
}
boolean retained = false;
try {
Chunk chunk = reader.read(connection, locator, 0, PREVIEW_CHARACTERS + 1);
if (chunk.eof()) return new Preview(chunk.data(), null);
String text = chunk.data();
int count = text.codePointCount(0, text.length());
if (count <= PREVIEW_CHARACTERS) return new Preview(text, null);
String ref = UUID.randomUUID().toString();
entries.put(ref, new Entry(connection, locator));
retained = true;
return new Preview(text.substring(0, text.offsetByCodePoints(0, PREVIEW_CHARACTERS)), ref);
} finally {
if (!retained) locator.free();
}
}
synchronized Chunk fetch(Connection connection, String ref, long offset, int limit) throws SQLException {
if (offset < 0 || offset == Long.MAX_VALUE || limit < 1 || limit > MAX_CHUNK_CHARACTERS) {
throw new IllegalArgumentException("Invalid LOB chunk offset or limit");
}
prune();
Entry entry = entries.get(ref);
if (entry == null || entry.connection != connection || connection.isClosed()) {
release(ref);
return new Chunk("expired", "", offset, false, "text");
}
entry.accessed = System.currentTimeMillis();
try {
return reader.read(connection, entry.locator, offset, limit);
} catch (SQLException error) {
release(ref);
throw error;
}
}
synchronized boolean release(String ref) {
Entry entry = entries.remove(ref);
if (entry == null) return false;
free(entry);
return true;
}
synchronized boolean hasValues() {
prune();
return !entries.isEmpty();
}
synchronized void clear() {
entries.values().forEach(OceanBaseLobValues::free);
entries.clear();
}
private void prune() {
long now = System.currentTimeMillis();
Iterator<Entry> iterator = entries.values().iterator();
while (iterator.hasNext()) {
Entry entry = iterator.next();
if (now - entry.accessed >= IDLE_MILLIS || now - entry.created >= MAX_AGE_MILLIS) {
iterator.remove();
free(entry);
}
}
}
private static void free(Entry entry) {
try { entry.locator.free(); } catch (SQLException ignored) { }
}
static Chunk read(Connection connection, Clob locator, long offset, int limit) throws SQLException {
try (CallableStatement call = connection.prepareCall("{call DBMS_LOB.READ(?, ?, ?, ?)}")) {
if (!(call instanceof OceanBaseStatement)) throw new SQLException("Unsupported OceanBase LOB statement");
((OceanBaseStatement) call).setInternal();
call.setClob(1, locator);
call.setInt(2, limit);
call.setLong(3, offset + 1);
call.registerOutParameter(2, Types.INTEGER);
call.registerOutParameter(4, Types.VARCHAR);
call.setQueryTimeout(20);
try {
JdbcExecutor.current().withActiveStatement(call, () -> { call.execute(); return null; });
} catch (SQLException error) {
// DBMS_LOB.READ signals an empty LOB or reading beyond its end with NO_DATA_FOUND.
if (error.getErrorCode() == 1403) return new Chunk("ok", "", offset, true, "text");
throw error;
}
int amount = call.getInt(2);
byte[] bytes = call.getBytes(4);
if (amount < 0 || amount > limit || bytes == null && amount != 0) {
throw new SQLException("Invalid OceanBase LOB chunk response");
}
String text;
try {
text = bytes == null ? "" : StandardCharsets.UTF_8.newDecoder()
.onMalformedInput(CodingErrorAction.REPORT).onUnmappableCharacter(CodingErrorAction.REPORT)
.decode(ByteBuffer.wrap(bytes)).toString();
} catch (java.nio.charset.CharacterCodingException error) {
throw new SQLException("Invalid UTF-8 in OceanBase LOB chunk", error);
}
if (text.codePointCount(0, text.length()) != amount) throw new SQLException("OceanBase LOB character count mismatch");
return new Chunk("ok", text, Math.addExact(offset, amount), amount < limit, "text");
}
}
static boolean isCharacterLob(int sqlType, String typeName) {
return sqlType == Types.CLOB || sqlType == Types.NCLOB
|| "CLOB".equalsIgnoreCase(typeName) || "NCLOB".equalsIgnoreCase(typeName);
}
private static boolean hasMarkerCollision(ResultSet rs) throws SQLException {
var metadata = rs.getMetaData();
for (int index = 1; index <= metadata.getColumnCount(); index++) {
if (metadata.getColumnLabel(index).toUpperCase(Locale.ROOT).startsWith(MARKER_PREFIX)) return true;
}
return false;
}
static MarkedRows mark(List<String> columns, List<String> types, List<List<Object>> rows) {
List<String> expandedColumns = new ArrayList<>();
List<String> expandedTypes = new ArrayList<>();
boolean[] marked = new boolean[columns.size()];
for (List<Object> row : rows) {
for (int index = 0; index < row.size(); index++) if (row.get(index) instanceof Preview) marked[index] = true;
}
boolean anyMarked = false;
for (boolean value : marked) anyMarked |= value;
if (!anyMarked) return null;
for (int index = 0; index < columns.size(); index++) {
expandedColumns.add(columns.get(index));
expandedTypes.add(index < types.size() ? types.get(index) : "");
if (marked[index]) {
expandedColumns.add(MARKER_PREFIX + "C_" + index);
expandedTypes.add("VARCHAR");
}
}
List<List<Object>> expandedRows = new ArrayList<>();
for (List<Object> row : rows) {
List<Object> expanded = new ArrayList<>();
for (int index = 0; index < columns.size(); index++) {
Object value = row.get(index);
Preview preview = value instanceof Preview ? (Preview) value : null;
expanded.add(preview == null ? value : preview.text());
if (marked[index]) expanded.add(preview == null || preview.ref() == null ? null : "D:1:unknown:" + preview.ref());
}
expandedRows.add(expanded);
}
return new MarkedRows(expandedColumns, expandedTypes, expandedRows);
}
}
@@ -28,6 +28,7 @@ import com.dbx.agent.QueryPageOptions;
import com.dbx.agent.QueryPageResult;
import com.dbx.agent.QueryResult;
import com.dbx.agent.QueryTiming;
import com.dbx.agent.SpatialColumn;
import com.dbx.agent.TableInfo;
import com.dbx.agent.TriggerInfo;
@@ -66,6 +67,8 @@ public final class OceanBaseOracleAgent extends ConfiguredJdbcAgent {
"GGSYS", "FLOWS_FILES", "APEX_PUBLIC_USER", "GSMROOTUSER", "SYSRAC"
);
private boolean queryTimeoutChanged;
private final OceanBaseLobValues lobValues = new OceanBaseLobValues();
private boolean deferCharacterLobs;
public static final JdbcAgentProfile OCEANBASE_ORACLE_PROFILE = new JdbcAgentProfile(
"com.oceanbase.jdbc.Driver",
@@ -91,15 +94,32 @@ public final class OceanBaseOracleAgent extends ConfiguredJdbcAgent {
@Override
public QueryResult executeQuery(String sql, String schema, ExecuteQueryOptions options) {
try (QueryTiming timing = QueryTiming.begin()) {
prepareLobExecution(sql, options.getDeferLobs());
QueryResult result = super.executeQuery(sql, schema, options);
var marked = OceanBaseLobValues.mark(result.getColumns(), result.getColumn_types(), result.getRows());
if (marked != null) {
var spatialColumns = remapLobSpatialColumns(result.getSpatial_columns(), marked.columns());
var spatialValues = remapLobSpatialValues(result.getSpatial_values(), marked.columns());
result.setColumns(marked.columns());
result.setColumn_types(marked.types());
result.setRows(marked.rows());
result.setSpatial_columns(spatialColumns);
result.setSpatial_values(spatialValues);
}
result.setQuery_timings_ms(timing.finish());
return result;
} catch (RuntimeException error) {
lobValues.clear();
throw error;
} finally {
deferCharacterLobs = false;
}
}
@Override
public QueryPageResult executeQueryPage(String sql, String schema, QueryPageOptions options) {
try (QueryTiming timing = QueryTiming.begin()) {
prepareLobExecution(sql, options.getDeferLobs());
long prepareStarted = System.nanoTime();
Connection connection = requireConnected();
uncheckedVoid(() -> beforeQueryExecution(connection, options.getTimeoutSecs()));
@@ -107,8 +127,14 @@ public final class OceanBaseOracleAgent extends ConfiguredJdbcAgent {
QueryPageResult result = JdbcExecutor.current().executeBoundedPage(
connection, sql, schema, this::setSchemaSQL, this::resetSchemaSQL, options, resultValueReader()
);
markLobPage(result);
result.setQuery_timings_ms(timing.finish());
return result;
} catch (RuntimeException error) {
lobValues.clear();
throw error;
} finally {
deferCharacterLobs = false;
}
}
@@ -116,11 +142,102 @@ public final class OceanBaseOracleAgent extends ConfiguredJdbcAgent {
public QueryPageResult fetchQueryPage(String sessionId, int pageSize) {
try (QueryTiming timing = QueryTiming.begin()) {
QueryPageResult result = super.fetchQueryPage(sessionId, pageSize);
markLobPage(result);
result.setQuery_timings_ms(timing.finish());
return result;
}
}
@Override
public QueryPageResult startTableRead(String sql, String schema, QueryPageOptions options) {
prepareLobExecution(sql, options.getDeferLobs());
try {
QueryPageResult result = super.startTableRead(sql, schema, options);
markLobPage(result);
return result;
} catch (RuntimeException error) {
lobValues.clear();
throw error;
} finally {
deferCharacterLobs = false;
}
}
@Override
public QueryPageResult fetchTableReadPage(String sessionId, int pageSize) {
QueryPageResult result = super.fetchTableReadPage(sessionId, pageSize);
markLobPage(result);
return result;
}
private static void markLobPage(QueryPageResult result) {
var marked = OceanBaseLobValues.mark(result.getColumns(), result.getColumn_types(), result.getRows());
if (marked == null) return;
var spatialColumns = remapLobSpatialColumns(result.getSpatial_columns(), marked.columns());
var spatialValues = remapLobSpatialValues(result.getSpatial_values(), marked.columns());
result.setColumns(marked.columns());
result.setColumn_types(marked.types());
result.setRows(marked.rows());
result.setSpatial_columns(spatialColumns);
result.setSpatial_values(spatialValues);
}
private static List<SpatialColumn> remapLobSpatialColumns(List<SpatialColumn> spatial, List<String> columns) {
List<Integer> retained = new ArrayList<>();
for (int index = 0; index < columns.size(); index++) {
if (!columns.get(index).startsWith(OceanBaseLobValues.MARKER_PREFIX)) retained.add(index);
}
List<SpatialColumn> remapped = new ArrayList<>();
for (SpatialColumn column : spatial) remapped.add(new SpatialColumn(retained.get(column.getColumn_index()), column.getSrid()));
return remapped;
}
private static List<List<Integer>> remapLobSpatialValues(List<List<Integer>> spatial, List<String> columns) {
List<List<Integer>> remapped = new ArrayList<>();
for (List<Integer> row : spatial) {
List<Integer> expanded = new ArrayList<>();
int sourceIndex = 0;
for (String column : columns) {
expanded.add(column.startsWith(OceanBaseLobValues.MARKER_PREFIX) ? null : row.get(sourceIndex++));
}
remapped.add(expanded);
}
return remapped;
}
private void prepareLobExecution(String sql, boolean requested) {
// Unproven statements may end the transaction or replace an object. Fail closed.
String leading = sql.replaceFirst("(?s)^(?:\\s|--[^\\r\\n]*(?:\\r?\\n|$)|/\\*.*?\\*/)*", "");
if (!leading.matches("(?is)^SELECT\\b.*")) lobValues.clear();
deferCharacterLobs = requested;
}
@Override
protected JdbcExecutor.ResultValueReader resultValueReader() {
if (!deferCharacterLobs) return super.resultValueReader();
return (JdbcExecutor.ColumnAwareResultValueReader) (rs, index, sqlType, typeName) -> {
Object preview = lobValues.preview(rs, index, sqlType, typeName);
return preview == null ? resultValue(rs, index, sqlType) : preview;
};
}
@Override
protected boolean hasRetainedResultResources() { return lobValues.hasValues(); }
@Override
protected void releaseRetainedResultResources() { lobValues.clear(); }
@Override
public void invalidateLargeValues() { lobValues.clear(); }
@Override
public Object readLargeValueChunk(String valueRef, long offset, int limit) {
return unchecked(() -> lobValues.fetch(requireConnected(), valueRef, offset, limit));
}
@Override
public boolean releaseLargeValue(String valueRef) { return lobValues.release(valueRef); }
@Override
protected String buildJdbcUrl(ConnectParams params) {
return buildUrl(params);
@@ -0,0 +1,67 @@
package com.dbx.agent.oceanbaseoracle;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Assumptions;
import static org.junit.jupiter.api.Assertions.*;
import java.sql.DriverManager;
import java.sql.Types;
import java.util.UUID;
/** Opt-in isolated Oracle tenant only. Run during the unified acceptance phase. */
class OceanBaseLobSnapshotIntegrationTest {
@Test
void capturesOriginalClobAcrossCursorCloseAndConcurrentUpdateAndDelete() throws Exception {
String url = System.getenv("OB_LOB_INTEGRATION_URL");
Assumptions.assumeTrue(url != null && !url.isBlank(), "Isolated OB LOB integration environment not configured");
String user = System.getenv("OB_LOB_INTEGRATION_USER");
String password = System.getenv("OB_LOB_INTEGRATION_PASSWORD");
String table = "DBX_O13_" + UUID.randomUUID().toString().replace("-", "").substring(0, 20).toUpperCase();
OceanBaseLobValues values = new OceanBaseLobValues();
try (var connection = DriverManager.getConnection(url, user, password);
var other = DriverManager.getConnection(url, user, password)) {
try (var statement = connection.createStatement(); var tenant = statement.executeQuery("SELECT SYS_CONTEXT('USERENV','CON_NAME') FROM DUAL")) {
assertTrue(tenant.next());
assertEquals("oracletest", tenant.getString(1).toLowerCase(), "Refuse writes outside the authorized isolated tenant");
}
boolean created = false;
try {
try (var statement = connection.createStatement()) {
statement.execute("CREATE TABLE " + table + " (ID NUMBER PRIMARY KEY, PAYLOAD CLOB)");
created = true;
}
String original = "中文😀完整原值".repeat(12_000);
try (var insert = connection.prepareStatement("INSERT INTO " + table + " VALUES (1, ?)")) {
insert.setString(1, original);
insert.executeUpdate();
}
OceanBaseLobValues.Preview preview;
try (var statement = connection.createStatement(); var result = statement.executeQuery("SELECT PAYLOAD FROM " + table + " WHERE ID = 1")) {
assertTrue(result.next());
preview = (OceanBaseLobValues.Preview) values.preview(result, 1, Types.CLOB, "CLOB");
}
assertNotNull(preview.ref());
assertEquals(256, preview.text().codePointCount(0, preview.text().length()));
try (var update = other.prepareStatement("UPDATE " + table + " SET PAYLOAD = ? WHERE ID = 1")) {
update.setString(1, "已经变更😀".repeat(12_000));
assertEquals(1, update.executeUpdate());
}
try (var statement = other.createStatement()) { assertEquals(1, statement.executeUpdate("DELETE FROM " + table + " WHERE ID = 1")); }
StringBuilder complete = new StringBuilder();
long offset = 0;
while (true) {
var chunk = values.fetch(connection, preview.ref(), offset, 4096);
assertEquals("ok", chunk.status());
complete.append(chunk.data());
if (chunk.eof()) break;
offset = chunk.next_offset();
}
assertEquals(original, complete.toString());
values.clear();
assertEquals("expired", values.fetch(connection, preview.ref(), 0, 4096).status());
} finally {
values.clear();
if (created) try (var statement = connection.createStatement()) { statement.execute("DROP TABLE " + table + " PURGE"); }
}
}
}
}
@@ -0,0 +1,140 @@
package com.dbx.agent.oceanbaseoracle;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
import java.lang.reflect.Proxy;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
class OceanBaseLobValuesTest {
@Test
void boundsPreviewAndReadsTheSameCapturedLocatorWithoutMaterializingItsStreams() throws Exception {
String original = "中文😀".repeat(1000);
List<Integer> amounts = new ArrayList<>();
Fixture fixture = new Fixture();
OceanBaseLobValues values = new OceanBaseLobValues((connection, locator, offset, limit) -> {
assertSame(fixture.connection, connection);
assertSame(fixture.clob, locator);
amounts.add(limit);
return chunk(original, offset, limit);
});
var preview = (OceanBaseLobValues.Preview) values.preview(fixture.resultSet, 1, Types.CLOB, "CLOB");
assertEquals(256, preview.text().codePointCount(0, preview.text().length()));
assertFalse(Character.isHighSurrogate(preview.text().charAt(preview.text().length() - 1)));
assertNotNull(preview.ref());
assertEquals(List.of(257), amounts);
StringBuilder complete = new StringBuilder();
long offset = 0;
while (true) {
var next = values.fetch(fixture.connection, preview.ref(), offset, 31);
complete.append(next.data());
if (next.eof()) break;
assertTrue(next.next_offset() > offset);
offset = next.next_offset();
}
assertEquals(original, complete.toString());
assertFalse(fixture.freed.get());
assertTrue(values.release(preview.ref()));
assertTrue(fixture.freed.get());
assertEquals("expired", values.fetch(fixture.connection, preview.ref(), 0, 31).status());
}
@Test
void distinguishesNullEmptyShortAndDeferredValues() throws Exception {
for (String text : List.of("", "短😀值", "x".repeat(257))) {
Fixture fixture = new Fixture();
OceanBaseLobValues values = new OceanBaseLobValues((connection, locator, offset, limit) -> chunk(text, offset, limit));
var preview = (OceanBaseLobValues.Preview) values.preview(fixture.resultSet, 1, Types.CLOB, "CLOB");
// A value exactly filling the look-ahead chunk remains deferred until EOF is known.
if (text.length() < 257) {
assertEquals(text, preview.text());
assertNull(preview.ref());
assertTrue(fixture.freed.get());
} else assertNotNull(preview.ref());
values.clear();
}
Fixture fixture = new Fixture();
fixture.nullValue = true;
OceanBaseLobValues values = new OceanBaseLobValues((connection, locator, offset, limit) -> { fail("NULL must not be read"); return null; });
var preview = (OceanBaseLobValues.Preview) values.preview(fixture.resultSet, 1, Types.CLOB, "CLOB");
assertNull(preview.text());
assertNull(preview.ref());
}
@Test
void rejectsWrongConnectionAndInvalidChunkBoundsAndInvalidatesOnClear() throws Exception {
Fixture fixture = new Fixture();
OceanBaseLobValues values = new OceanBaseLobValues((connection, locator, offset, limit) -> chunk("原值".repeat(1000), offset, limit));
var preview = (OceanBaseLobValues.Preview) values.preview(fixture.resultSet, 1, Types.CLOB, "CLOB");
assertThrows(IllegalArgumentException.class, () -> values.fetch(fixture.connection, preview.ref(), -1, 1));
assertThrows(IllegalArgumentException.class, () -> values.fetch(fixture.connection, preview.ref(), 0, 4097));
assertEquals("expired", values.fetch(new Fixture().connection, preview.ref(), 0, 1).status());
Fixture nextFixture = new Fixture();
var nextPreview = (OceanBaseLobValues.Preview) values.preview(nextFixture.resultSet, 1, Types.CLOB, "CLOB");
values.clear();
assertEquals("expired", values.fetch(nextFixture.connection, nextPreview.ref(), 0, 1).status());
assertTrue(fixture.freed.get());
assertTrue(nextFixture.freed.get());
}
@Test
void markerCollisionFallsBackBeforeReadingOrTruncatingAndRefsStayOutOfValues() throws Exception {
Fixture fixture = new Fixture();
fixture.label = "__dbx_large_value_bytes_c_0";
OceanBaseLobValues values = new OceanBaseLobValues((connection, locator, offset, limit) -> { fail("Collision must keep complete ordinary read"); return null; });
assertNull(values.preview(fixture.resultSet, 1, Types.CLOB, "CLOB"));
var marked = OceanBaseLobValues.mark(List.of("ID", "Payload"), List.of("NUMBER", "CLOB"),
List.of(List.of(1, new OceanBaseLobValues.Preview("原始😀", "opaque-ref")),
java.util.Arrays.asList(2, new OceanBaseLobValues.Preview(null, null))));
assertEquals(List.of("ID", "Payload", "__DBX_LARGE_VALUE_BYTES_C_1"), marked.columns());
assertEquals("原始😀", marked.rows().get(0).get(1));
assertEquals("D:1:unknown:opaque-ref", marked.rows().get(0).get(2));
assertNull(marked.rows().get(1).get(1));
assertNull(marked.rows().get(1).get(2));
}
private static OceanBaseLobValues.Chunk chunk(String text, long offset, int limit) {
int count = text.codePointCount(0, text.length());
int start = Math.min(count, (int) offset);
int end = Math.min(count, start + limit);
String data = text.substring(text.offsetByCodePoints(0, start), text.offsetByCodePoints(0, end));
return new OceanBaseLobValues.Chunk("ok", data, offset + end - start, end - start < limit, "text");
}
private static final class Fixture {
final AtomicBoolean freed = new AtomicBoolean();
boolean nullValue;
String label = "PAYLOAD";
final Connection connection = proxy(Connection.class, (method, args) -> method.equals("isClosed") ? false : null);
final Clob clob = proxy(Clob.class, (method, args) -> {
if (method.equals("free")) { freed.set(true); return null; }
throw new AssertionError("Unexpected full LOB API: " + method);
});
final Statement statement = proxy(Statement.class, (method, args) -> method.equals("getConnection") ? connection : null);
final ResultSetMetaData metadata = proxy(ResultSetMetaData.class, (method, args) -> {
if (method.equals("getColumnCount")) return 1;
if (method.equals("getColumnLabel")) return label;
return null;
});
final ResultSet resultSet = proxy(ResultSet.class, (method, args) -> {
if (method.equals("getMetaData")) return metadata;
if (method.equals("getStatement")) return statement;
if (method.equals("getClob")) return nullValue ? null : clob;
if (method.equals("wasNull")) return nullValue;
throw new AssertionError("Unexpected result read: " + method);
});
}
@FunctionalInterface interface Handler { Object call(String method, Object[] args) throws Throwable; }
private static <T> T proxy(Class<T> type, Handler handler) {
return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type},
(object, method, args) -> handler.call(method.getName(), args)));
}
}
@@ -333,6 +333,7 @@ import { translateBackendError } from "@/i18n/backend-errors";
import { useNavigationTargets } from "@/composables/useNavigationTargets";
import { useDataGridExport, type MongoCopyUpdateTarget } from "@/composables/useDataGridExport";
import { isTauriRuntime } from "@/lib/backend/tauriRuntime";
import { promptExportSavePath } from "@/lib/export/exportPath";
import { eventTargetAllowsNativeClipboard, isPlainClipboardShortcut, readTextFromClipboard } from "@/lib/common/clipboard";
import { claimDataGridPaste, claimDataGridSelectAll, clearDataGridClipboardCopy, parseDataGridClipboard, parseDataGridClipboardInBatches, planDataGridPaste } from "@/lib/dataGrid/dataGridClipboard";
import { parseInsertStatementPaste, parseInsertStatementPasteInBatches } from "@/lib/dataGrid/dataGridInsertPaste";
@@ -10619,6 +10620,23 @@ async function downloadDetailBinaryValue(detail: DataGridCellDetail | null, mode
}
function binaryDownloadSubmenu(detail: DataGridCellDetail | null): ContextMenuItem | null {
if (detail && largeValueRuntimeInstance.snapshotReference(getRowItem(detail.rowId), detail.colIndex)
&& !isBinaryCellColumnType(detail.type)) {
return {
label: t("grid.downloadSnapshotValue"), icon: Download,
action: async () => {
const defaultFileName = `lob-row-${detail.rowNumber}.txt`;
const path = isTauriRuntime() ? await promptExportSavePath({ defaultFileName, filters: [{ name: "Text", extensions: ["txt"] }] }) : defaultFileName;
if (!path) return;
try {
await largeValueRuntimeInstance.downloadSnapshotCell(detail.rowId, detail.colIndex, path);
toast(isTauriRuntime() ? t("grid.downloadSaved", { path }) : t("grid.downloadStarted", { fileName: defaultFileName }));
} catch (error) {
reportLargeValueLoadError(error);
}
},
};
}
if (!canDownloadDetailBinaryValue(detail)) return null;
return {
label: t("grid.downloadBinaryValue"),
@@ -13,6 +13,7 @@ import { useToast } from "@/composables/useToast";
import { useExportTracker } from "@/composables/useExportTracker";
import { clipboardCellValue, type CellValue } from "@/lib/dataGrid/cellValue";
import { binaryCellClipboardText } from "@/lib/dataGrid/binaryCellDownload";
import { materializeSnapshotResultRows } from "@/lib/dataGrid/largeValueSnapshot";
import { tryStartExclusiveActivation, type ActionActivationGuard } from "@/lib/connection/actionActivation";
import { clipboardLineEndings, copyToClipboard } from "@/lib/common/clipboard";
import { clearDataGridClipboardCopy, rememberDataGridClipboardCopy } from "@/lib/dataGrid/dataGridClipboard";
@@ -1370,15 +1371,20 @@ export function useDataGridExport(options: UseDataGridExportOptions) {
const exportPattern = useSettingsStore().editorSettings.globalDateTimeExportFormat;
const rightAlign = useSettingsStore().editorSettings.numericColumnRightAlign;
const worksheets = sheets.map((sheet) => ({
const worksheets = [];
for (const sheet of sheets) {
const sourceRows = sheet.result.rows;
const rows = await materializeSnapshotResultRows(sheet.result, () => sheet.result.rows === sourceRows);
worksheets.push({
sheetName: sheet.sheetName,
columns: sheet.result.columns,
columnTypes: sheet.result.column_types ?? [],
columnComments: buildXlsxHeaderOverrides(sheet.result.columns, commentsForExportColumns(sheet.result.columns), exportOptions.headerMode),
rows: formatTemporalRowsForExport(sheet.result.rows, sheet.result.column_types ?? [], exportPattern),
rows: formatTemporalRowsForExport(rows, sheet.result.column_types ?? [], exportPattern),
numericColumnRightAlign: rightAlign,
autoFilter: exportOptions.autoFilter,
}));
});
}
const sqlWorksheet = includeSqlSheet ? buildXlsxSqlWorksheet(sheets.map((sheet) => ({ resultName: sheet.sheetName, sql: sheet.sql || sheet.result.sourceStatement || "" }))) : undefined;
await api.exportQueryResultsXlsx(outputPath, sqlWorksheet ? [...worksheets, { ...sqlWorksheet, autoFilter: false }] : worksheets, exportOptions.autoFilter, exportPattern || undefined);
notifyExportSuccess(outputPath);
@@ -5,6 +5,7 @@ import { buildDataGridContextFilterCondition } from "@/lib/dataGrid/dataGridSql"
import { buildTableSelectSql } from "@/lib/table/tableSelectSql";
import { shouldIncludeSyntheticRowId } from "@/lib/table/tableEditing";
import { queryTimeoutSecsForConnection } from "@/lib/sql/queryTimeout";
import { materializeLargeValueSnapshot } from "@/lib/dataGrid/largeValueSnapshot";
import type { ColumnInfo, ConnectionConfig, DatabaseType, QueryResult } from "@/types/database";
import type { DataGridRuntimeScope } from "@/lib/dataGrid/dataGridRuntime";
import {
@@ -90,6 +91,13 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
const LARGE_VALUE_FETCH_MAX_ROWS = 200;
const LARGE_VALUE_FETCH_TARGET_BYTES = 64 * 1024 * 1024;
const pendingLargeValueHydrations = createResultScopedPendingRequests<boolean>();
const snapshotExecutionIds = new Set<string>();
const cancelSnapshotReads = () => {
for (const executionId of snapshotExecutionIds) void api.cancelQuery(executionId).catch(() => {});
snapshotExecutionIds.clear();
};
watch(() => [options.result.value, options.connectionId.value, options.executionDatabase.value], cancelSnapshotReads);
options.runtimeScope.addCleanup(cancelSnapshotReads);
const largeValueCellsByKey = computed(() => largeValueCellMap(options.result.value));
type VisibleLargeValuePreviewRequest = {
item: LargeValueRowItem;
@@ -119,7 +127,28 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
function formatGridItemCell(item: LargeValueRowItem, columnIndex: number): string {
void visibleLargeValuePreviewVersion.value;
return options.formatCellCached(visibleLargeValuePreviewValue(item, columnIndex, item.data[columnIndex] ?? null), columnIndex, largeValueOriginalBytes(item, columnIndex));
const text = options.formatCellCached(visibleLargeValuePreviewValue(item, columnIndex, item.data[columnIndex] ?? null), columnIndex, largeValueOriginalBytes(item, columnIndex));
return snapshotReference(item, columnIndex) ? options.translate("grid.largeValueSnapshotPreview", { value: text }) : text;
}
function snapshotReference(item: LargeValueRowItem | undefined, columnIndex: number): string | undefined {
if (!item || !isLargeValuePreview(item, columnIndex) || item.sourceIndex === undefined) return undefined;
return largeValueCellsByKey.value.get(largeValueCellKey(item.sourceIndex, columnIndex))?.value_ref;
}
async function downloadSnapshotCell(rowId: number, columnIndex: number, filePath: string): Promise<void> {
const sourceResult = options.result.value;
const ref = snapshotReference(options.getRowItem(rowId), columnIndex);
const context = sourceResult.large_value_context;
if (!ref || !context) throw new Error("LOB snapshot is unavailable");
const executionId = options.uuid();
snapshotExecutionIds.add(executionId);
try {
await api.downloadLargeValue({ ...context, valueRef: ref, executionId }, filePath);
if (options.result.value !== sourceResult) throw new Error("LOB result context changed");
} finally {
snapshotExecutionIds.delete(executionId);
}
}
function formatGridItemCellForConfirmation(item: LargeValueRowItem, columnIndex: number): string {
@@ -477,6 +506,12 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
}
async function resolveLargeValueCells(rowIds: number[], columnIndexes: number[]): Promise<ResolvedLargeValueCells> {
const sourceResult = options.result.value;
const connectionId = options.connectionId.value;
const executionDatabase = options.executionDatabase.value;
const operation = options.resultLifecycle.beginOperation();
const isCurrent = () => options.result.value === sourceResult && options.connectionId.value === connectionId
&& options.executionDatabase.value === executionDatabase && options.resultLifecycle.isCurrent(operation);
const resolved: ResolvedLargeValueCells = new Map();
const requestedColumns = new Set(columnIndexes);
const requestsByColumn = new Map<number, LargeValueCellRequest[]>();
@@ -487,6 +522,23 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
if (!isLargeValuePreview(item, columnIndex)) continue;
const metadata = largeValueCellsByKey.value.get(largeValueCellKey(item.sourceIndex, columnIndex));
if (!metadata) continue;
if (metadata.value_ref) {
const context = sourceResult.large_value_context;
if (!context || context.connectionId !== connectionId || context.database !== executionDatabase) {
throw new Error("LOB result connection is unavailable; execute the query again");
}
const executionId = options.uuid();
snapshotExecutionIds.add(executionId);
try {
const value = await materializeLargeValueSnapshot({ ...context, valueRef: metadata.value_ref, executionId }, isCurrent);
const rowValues = resolved.get(item.id) ?? new Map<number, CellValue>();
rowValues.set(columnIndex, value);
resolved.set(item.id, rowValues);
} finally {
snapshotExecutionIds.delete(executionId);
}
continue;
}
const requests = requestsByColumn.get(columnIndex) ?? [];
requests.push({ item, sourceIndex: item.sourceIndex, columnIndex, originalBytes: metadata.original_bytes });
requestsByColumn.set(columnIndex, requests);
@@ -507,6 +559,7 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
for (const [columnIndex, requests] of requestsByColumn) {
for (const chunk of chunkLargeValueRequests(requests)) {
await fetchLargeValueRequestChunk(columnIndex, chunk, primaryKeyIndexes, resolved);
if (!isCurrent()) throw new Error("LOB result context changed");
}
}
return resolved;
@@ -557,6 +610,11 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
const rows = sourceResult.rows.slice();
rows[item.sourceIndex!] = row;
sourceResult.rows = rows;
const snapshotRef = sourceResult.large_value_cells?.find((cell) => cell.row_index === item.sourceIndex && cell.column_index === columnIndex)?.value_ref;
if (snapshotRef && sourceResult.large_value_context) {
void api.releaseLargeValue({ ...sourceResult.large_value_context, valueRef: snapshotRef })
.catch((error) => options.appendDebugLog("warn", "[DBX][DataGrid:large-value] release failed", error));
}
visibleLargeValuePreviewCaches.get(sourceResult)?.forget(item.sourceIndex!, columnIndex);
sourceResult.large_value_cells = sourceResult.large_value_cells?.filter((cell) => cell.row_index !== item.sourceIndex || cell.column_index !== columnIndex);
options.largeValueResolutionVersion.value += 1;
@@ -571,6 +629,8 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) {
}
return {
snapshotReference,
downloadSnapshotCell,
isLargeValuePreview,
largeValueOriginalBytes,
formatGridItemCell,
+2
View File
@@ -2862,6 +2862,8 @@ export default {
binaryImportFailed: "Failed to import file: {message}",
binaryImportTooLarge: "File is too large ({size}, limit {limit}). Importing large files into a single cell is not supported.",
downloadBinaryValue: "Download Value",
downloadSnapshotValue: "Download complete result snapshot",
largeValueSnapshotPreview: "{value} (result snapshot preview)",
downloadSaved: "Saved to {path}",
downloadStarted: "Download started: {fileName}",
binaryDownload: {
+2
View File
@@ -2797,6 +2797,8 @@ export default withEnglishFallback({
binaryImportFailed: "导入文件失败:{message}",
binaryImportTooLarge: "文件过大({size},上限 {limit})。不支持将大文件导入到单个单元格。",
downloadBinaryValue: "下载值",
downloadSnapshotValue: "下载完整结果快照",
largeValueSnapshotPreview: "{value}(结果快照预览)",
downloadSaved: "已保存到 {path}",
downloadStarted: "已交给浏览器下载:{fileName}",
binaryDownload: {
+2
View File
@@ -2572,6 +2572,8 @@ export default withEnglishFallback({
binaryImportFailed: "匯入檔案失敗:{message}",
binaryImportTooLarge: "檔案過大({size},上限 {limit})。不支援將大檔案匯入到單一儲存格。",
downloadBinaryValue: "下載值",
downloadSnapshotValue: "下載完整結果快照",
largeValueSnapshotPreview: "{value}(結果快照預覽)",
downloadSaved: "已儲存到 {path}",
downloadStarted: "已交給瀏覽器下載:{fileName}",
binaryDownload: {
@@ -0,0 +1,64 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
const mocks = vi.hoisted(() => ({ readLargeValueChunk: vi.fn() }));
vi.mock("@/lib/backend/api", () => mocks);
import { materializeLargeValueSnapshot, materializeSnapshotResultRows, readLargeValueSnapshot } from "@/lib/dataGrid/largeValueSnapshot";
const request = { connectionId: "ob", database: "APP", clientSessionId: "tab-original", valueRef: "original-ref" };
const response = (data: string, next_offset: number, eof = false) => ({ status: "ok", data, next_offset, eof, value_kind: "text" });
beforeEach(() => mocks.readLargeValueChunk.mockReset());
describe("original-result LOB snapshot consumption", () => {
it("uses server character offsets and preserves Chinese and emoji across chunks", async () => {
mocks.readLargeValueChunk.mockResolvedValueOnce(response("中文😀", 3)).mockResolvedValueOnce(response("末尾😀", 6, true));
expect(await materializeLargeValueSnapshot(request, () => true)).toBe("中文😀末尾😀");
expect(mocks.readLargeValueChunk.mock.calls.map(([args]) => args.offset)).toEqual([0, 3]);
expect(mocks.readLargeValueChunk.mock.calls[1]?.[0]).toMatchObject({ valueRef: "original-ref", clientSessionId: "tab-original" });
});
it("refuses an expired original locator without returning a preview or re-querying a row", async () => {
mocks.readLargeValueChunk.mockResolvedValue({ ...response("", 0), status: "expired" });
await expect(materializeLargeValueSnapshot(request, () => true)).rejects.toThrow("expired");
expect(mocks.readLargeValueChunk).toHaveBeenCalledTimes(1);
});
it("rejects changed context after await before any consumer sees the value", async () => {
let current = true;
mocks.readLargeValueChunk.mockImplementation(async () => { current = false; return response("wrong context", 1, true); });
const consume = vi.fn();
await expect(readLargeValueSnapshot(request, () => current, consume)).rejects.toThrow("context changed");
expect(consume).not.toHaveBeenCalled();
});
it("rejects non-progressing chunks instead of looping or copying partial data", async () => {
mocks.readLargeValueChunk.mockResolvedValue(response("prefix", 0));
await expect(materializeLargeValueSnapshot(request, () => true)).rejects.toThrow("offset");
expect(mocks.readLargeValueChunk).toHaveBeenCalledTimes(1);
});
it("rejects expiry between chunks without returning the already fetched prefix", async () => {
mocks.readLargeValueChunk.mockResolvedValueOnce(response("前缀😀", 3))
.mockResolvedValueOnce({ ...response("", 3), status: "expired" });
await expect(materializeLargeValueSnapshot(request, () => true)).rejects.toThrow("expired");
expect(mocks.readLargeValueChunk).toHaveBeenCalledTimes(2);
});
it("validates character offsets rather than UTF-16 length", async () => {
mocks.readLargeValueChunk.mockResolvedValue(response("😀", 2, true));
await expect(materializeLargeValueSnapshot(request, () => true)).rejects.toThrow("character count");
});
it("uses each export result's original connection and column identity and leaves previews untouched", async () => {
const result = {
columns: ["Payload", "PAYLOAD"], column_types: ["CLOB", "CLOB"], rows: [["a-preview", "b-preview"]],
affected_rows: 0, execution_time_ms: 0,
large_value_cells: [{ row_index: 0, column_index: 1, original_bytes: 0, value_ref: "uppercase-column-ref" }],
large_value_context: { connectionId: "other-ob", database: "OTHER", txnSessionId: "original-transaction" },
};
mocks.readLargeValueChunk.mockResolvedValue(response("完整😀", 3, true));
expect(await materializeSnapshotResultRows(result, () => true)).toEqual([["a-preview", "完整😀"]]);
expect(result.rows).toEqual([["a-preview", "b-preview"]]);
expect(mocks.readLargeValueChunk).toHaveBeenCalledWith(expect.objectContaining({
connectionId: "other-ob", database: "OTHER", txnSessionId: "original-transaction", valueRef: "uppercase-column-ref",
}));
});
});
+3
View File
@@ -281,6 +281,9 @@ export const cancelQuery = forward("cancelQuery");
export const cancelQueryAndWait = forward("cancelQueryAndWait");
export const cancelConditionalUpdate = forward("cancelConditionalUpdate");
export const closeQuerySession = forward("closeQuerySession");
export const readLargeValueChunk = forward("readLargeValueChunk");
export const releaseLargeValue = forward("releaseLargeValue");
export const downloadLargeValue = forward("downloadLargeValue");
export const closeClientConnectionSession = forward("closeClientConnectionSession");
export const analyzeSqlReferences = forward("analyzeSqlReferences");
export const findStatementAtCursor = forward("findStatementAtCursor");
+37
View File
@@ -1587,6 +1587,43 @@ export async function executeMultiWithProgress(
return results;
}
export interface LargeValueRequest {
connectionId: string;
database: string;
valueRef: string;
offset?: number;
limit?: number;
executionId?: string;
clientSessionId?: string;
catalog?: string;
txnSessionId?: string;
}
export interface LargeValueChunk {
status: string;
data: string;
next_offset: number;
eof: boolean;
value_kind: "text" | "binary";
}
export async function readLargeValueChunk(request: LargeValueRequest): Promise<LargeValueChunk> {
return post("/api/query/large-value/chunk", request);
}
export async function releaseLargeValue(request: LargeValueRequest): Promise<boolean> {
return post("/api/query/large-value/release", request);
}
export async function downloadLargeValue(request: LargeValueRequest, filePath: string): Promise<void> {
const fileName = filePath.split(/[\\/]/).pop() || "lob.txt";
const prepared = await post<{ downloadId: string }>("/api/query/large-value/download", { request, fileName });
const anchor = document.createElement("a");
anchor.href = apiUrl(`/api/query/large-value/download/${encodeURIComponent(prepared.downloadId)}`);
anchor.download = fileName;
anchor.click();
}
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string, catalog?: string): Promise<boolean> {
return post("/api/query/close-session", {
connectionId,
+12
View File
@@ -2024,6 +2024,18 @@ export async function cancelConditionalUpdate(executionId: string): Promise<Cond
return invoke("cancel_conditional_update", { executionId });
}
export async function readLargeValueChunk(request: import("./http").LargeValueRequest): Promise<import("./http").LargeValueChunk> {
return invoke("read_large_value_chunk", { request });
}
export async function releaseLargeValue(request: import("./http").LargeValueRequest): Promise<boolean> {
return invoke("release_large_value", { request });
}
export async function downloadLargeValue(request: import("./http").LargeValueRequest, filePath: string): Promise<void> {
await invoke("download_large_value", { request, filePath });
}
export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string, catalog?: string): Promise<boolean> {
return invoke("close_query_session", {
connectionId,
@@ -0,0 +1,61 @@
import * as api from "@/lib/backend/api";
import type { LargeValueRequest, LargeValueChunk } from "@/lib/backend/http";
import type { QueryResult } from "@/types/database";
const MAX_MATERIALIZED_BYTES = 16 * 1024 * 1024;
/** No cache: every request validates the original server locator's lifecycle. */
export async function readLargeValueSnapshot(
request: LargeValueRequest,
isCurrent: () => boolean,
consume: (chunk: LargeValueChunk) => Promise<void> | void,
): Promise<void> {
let offset = 0;
while (true) {
if (!isCurrent()) throw new Error("LOB result context changed");
const chunk = await api.readLargeValueChunk({ ...request, offset, limit: 4096 });
if (!isCurrent()) throw new Error("LOB result context changed");
if (chunk.status !== "ok") throw new Error(`LOB snapshot ${chunk.status}; execute the query again`);
if (!Number.isSafeInteger(chunk.next_offset) || chunk.next_offset < offset || (!chunk.eof && chunk.next_offset === offset)) {
throw new Error("Invalid LOB chunk offset");
}
const amount = chunk.next_offset - offset;
if (amount > 4096 || chunk.value_kind === "text" && Array.from(chunk.data).length !== amount) {
throw new Error("Invalid LOB chunk character count");
}
await consume(chunk);
if (!isCurrent()) throw new Error("LOB result context changed");
if (chunk.eof) return;
offset = chunk.next_offset;
}
}
export async function materializeLargeValueSnapshot(request: LargeValueRequest, isCurrent: () => boolean): Promise<string> {
const chunks: string[] = [];
let bytes = 0;
await readLargeValueSnapshot(request, isCurrent, (chunk) => {
bytes += chunk.value_kind === "binary" ? chunk.data.length / 2 : new TextEncoder().encode(chunk.data).length;
if (bytes > MAX_MATERIALIZED_BYTES) throw new Error("LOB exceeds the 16 MiB view/copy limit; download the complete value instead");
chunks.push(chunk.data);
});
return chunks.join("");
}
export async function materializeSnapshotResultRows(result: QueryResult, isCurrent: () => boolean): Promise<QueryResult["rows"]> {
const cells = result.large_value_cells ?? [];
if (!cells.some((cell) => cell.value_ref)) return result.rows;
const context = result.large_value_context;
if (!context) throw new Error("LOB result connection is unavailable; execute the query again");
const rows = result.rows.map((row) => [...row]);
let totalBytes = 0;
for (const cell of cells) {
if (!cell.value_ref) throw new Error("This result includes a value without a snapshot reference");
const value = await materializeLargeValueSnapshot({ ...context, valueRef: cell.value_ref }, isCurrent);
totalBytes += new TextEncoder().encode(value).length;
if (totalBytes > 64 * 1024 * 1024) throw new Error("LOB result exceeds the 64 MiB export limit; download individual complete values");
const row = rows[cell.row_index];
if (!row || cell.column_index >= row.length) throw new Error("LOB result column is unavailable");
row[cell.column_index] = value;
}
return rows;
}
+45 -3
View File
@@ -847,7 +847,7 @@ function projectsAllColumnsForSource(analysis: EditableQueryInfo, sourceKey: str
}
function queryProjectsDeferredLob(databaseType: DatabaseType, analysis: EditableQueryInfo, sourceKey: string, columns: readonly { name: string; data_type: string }[]): boolean {
const deferredTypes = databaseType === "oracle" ? ORACLE_DEFERRED_LOB_TYPES : databaseType === "db2" ? DB2_DEFERRED_LOB_TYPES : undefined;
const deferredTypes = databaseType === "oracle" || databaseType === "oceanbase-oracle" ? ORACLE_DEFERRED_LOB_TYPES : databaseType === "db2" ? DB2_DEFERRED_LOB_TYPES : undefined;
if (!deferredTypes) return false;
const deferredColumns = new Set(
columns
@@ -2063,6 +2063,7 @@ export const useQueryStore = defineStore("query", () => {
if (!tab || !tab.resultRuns || runIndex < 0) return false;
const removedRun = tab.resultRuns[runIndex];
if (removedRun) void releaseResultLargeValues(removedRun.result, ...(removedRun.results ?? []));
if (removedRun?.resultSessionId) void closeResultRunSession(tab, removedRun);
if (removedRun?.resultCacheKey) void deleteTabResultSnapshot(removedRun.resultCacheKey);
if (removedRun) clearResultRunPayload(removedRun);
@@ -2092,6 +2093,7 @@ export const useQueryStore = defineStore("query", () => {
if (!tab.result && !tab.results?.length && !tab.resultEvicted) return false;
const closeSession = closeResultSession(tab);
await releaseResultLargeValues(tab.result, ...(tab.results ?? []));
releaseTabResultObjectPayloads(tab);
clearResultPayload(tab);
await closeSession;
@@ -2109,8 +2111,10 @@ export const useQueryStore = defineStore("query", () => {
const currentSessionId = tab.resultSessionId ?? tab.result?.session_id;
if (currentSessionId) closedSessionIds.add(currentSessionId);
const closeOperations = [closeResultSession(tab)];
closeOperations.push(releaseResultLargeValues(tab.result, ...(tab.results ?? [])));
for (const run of resultRuns) {
closeOperations.push(releaseResultLargeValues(run.result, ...(run.results ?? [])));
if (run.resultCacheKey) void deleteTabResultSnapshot(run.resultCacheKey);
if (!run.resultSessionId || closedSessionIds.has(run.resultSessionId)) continue;
closedSessionIds.add(run.resultSessionId);
@@ -2128,6 +2132,23 @@ export const useQueryStore = defineStore("query", () => {
return (tab.resultRuns?.reduce((max, run) => Math.max(max, run.sequence), 0) ?? 0) + 1;
}
async function releaseResultLargeValues(...results: Array<QueryResult | undefined>) {
const released = new Set<string>();
for (const result of results) {
const context = result?.large_value_context;
if (!context) continue;
for (const cell of result.large_value_cells ?? []) {
if (!cell.value_ref || released.has(cell.value_ref)) continue;
released.add(cell.value_ref);
try {
await api.releaseLargeValue({ ...context, valueRef: cell.value_ref });
} catch (error) {
console.warn("[DBX][large-value:release:error]", error);
}
}
}
}
async function closeResultRunSession(tab: QueryTab, run: NonNullable<QueryTab["resultRuns"]>[number]) {
if (!run.resultSessionId) return;
const target = run.multiDbExecution?.target;
@@ -6388,8 +6409,8 @@ export const useQueryStore = defineStore("query", () => {
function buildHiddenPrimaryKeyPreparation(tab: QueryTab, sql: string, databaseType: DatabaseType, loaded: LoadedEditableSource, primaryKeys: string[], declaredPrimaryKeys: string[], traceId: string, elapsed: () => string): EditableQueryExecutionPreparation {
const metadataAnalysis = expandStarProjectionColumnsForSource(bindColumnsForSource(databaseType, loaded.analysis, loaded.source, loaded.tableMeta.columns), loaded.source, loaded.tableMeta.columns);
const stableLobSource = databaseType === "oracle" ? oracleRowIdIsSafeForQuery(tab, loaded) : databaseType === "db2";
const largeValuePreview = (databaseType === "oracle" || databaseType === "db2") && primaryKeys.length > 0 && stableLobSource && columnsAllowDeferredLobMarkers(loaded.tableMeta.columns) && queryProjectsDeferredLob(databaseType, metadataAnalysis, loaded.source.key, loaded.tableMeta.columns);
const stableLobSource = databaseType === "oracle" || databaseType === "oceanbase-oracle" ? oracleRowIdIsSafeForQuery(tab, loaded) : databaseType === "db2";
const largeValuePreview = (databaseType === "oracle" || databaseType === "oceanbase-oracle" || databaseType === "db2") && primaryKeys.length > 0 && stableLobSource && columnsAllowDeferredLobMarkers(loaded.tableMeta.columns) && queryProjectsDeferredLob(databaseType, metadataAnalysis, loaded.source.key, loaded.tableMeta.columns);
const unchanged = { sql, metadataSql: sql, hiddenPrimaryKeys: [], largeValuePreview };
const missingPrimaryKeys =
declaredPrimaryKeys.length === 0
@@ -7962,6 +7983,12 @@ export const useQueryStore = defineStore("query", () => {
// would return the first page again (#8993).
const isOffsetJumpPage = typeof pageOffset === "number" && pageOffset > 0 && !options?.pagination?.sessionId;
const frontendTimeoutSecs = frontendQueryTimeoutSecsForSql(sqlToExecute, effectiveDbType, queryTimeoutSecs, sqlStatementParameterOptions);
if (tab.mode === "data" && effectiveDbType === "oceanbase-oracle") {
const meta = tableMetaForDataTab(tab);
useLargeValuePreview = !!meta && /^(?:BASE )?TABLE$/i.test(meta.tableType ?? "")
&& columnsAllowDeferredLobMarkers(meta.columns)
&& meta.columns.some((column) => /^(N?CLOB|BLOB)$/i.test(column.data_type.trim()));
}
const sourceLabelDatabase = targetDatabase || conn?.database;
const executionClientSessionId = options?.pagination?.clientSessionId ?? (tab.mode === "query" || tab.mode === "data" ? tabClientSessionId(tab) : undefined);
const currentBeforeDispatch = findExecutionTab(id);
@@ -8177,6 +8204,21 @@ export const useQueryStore = defineStore("query", () => {
const responseResults = await withFrontendQueryTimeout(executionPromise, frontendTimeoutSecs, t("editor.queryTimeoutError", { seconds: frontendTimeoutSecs }), () => {
void api.cancelQuery(executionId).catch((error) => queryExecutionLog("warn", "frontend-timeout:cancel-failed", { traceId, error }));
});
for (const result of responseResults) {
if (effectiveDbType === "oceanbase-oracle" && !useLargeValuePreview && result.column_types?.some((type) => /^(N?CLOB|BLOB)$/i.test(type.trim()))) {
result.messages = [...(result.messages ?? []), {
severity: "INFO", code: "LOB_COMPLETE_READ_FALLBACK",
message: "LOB values were read completely because this result does not have a verified single-table preview source.",
}];
}
if (result.large_value_cells?.some((cell) => cell.value_ref)) {
result.large_value_context = {
connectionId: executionConnectionId, database: executionDatabase,
clientSessionId: executionClientSessionId, txnSessionId: tab.autoCommit === false ? tab.txnSessionId : undefined,
catalog: executionCatalog,
};
}
}
if (findExecutionTab(id) !== tab || tab.executionId !== executionId || manualTransactionTargetEpoch(tab) !== executionTargetEpoch) return false;
// A single result has an unambiguous request boundary. This includes fetch and
// transport, but excludes SQL preparation and the grid's later render work.
+3 -1
View File
@@ -1391,7 +1391,9 @@ export interface QueryResult {
total_is_exact?: boolean;
truncated?: boolean;
/** Variable-length cells represented by bounded previews in `rows`. */
large_value_cells?: Array<{ row_index: number; column_index: number; original_bytes: number }>;
large_value_cells?: Array<{ row_index: number; column_index: number; original_bytes: number; value_ref?: string }>;
/** Frontend-only original result connection, independent of the currently active tab. */
large_value_context?: { connectionId: string; database: string; clientSessionId?: string; txnSessionId?: string; catalog?: string };
session_id?: string | null;
has_more?: boolean;
/** For Elasticsearch REST search results parsed into a _source table,
+180 -28
View File
@@ -482,10 +482,11 @@ struct ServerLargeValueMarker {
}
#[derive(Clone, Copy)]
struct ServerLargeValueMarkerValue {
struct ServerLargeValueMarkerValue<'a> {
kind: ServerLargeValuePreviewKind,
preview_size: usize,
original_bytes: Option<usize>,
value_ref: Option<&'a str>,
}
/// Some PostgreSQL-compatible servers (for example KingbaseES instances with
@@ -527,7 +528,7 @@ fn server_large_value_alias(
Some((source_index, Some(preview_kind), source_type))
}
fn server_large_value_marker(value: &serde_json::Value) -> Option<ServerLargeValueMarkerValue> {
fn server_large_value_marker(value: &serde_json::Value) -> Option<ServerLargeValueMarkerValue<'_>> {
let mut parts = value.as_str()?.split(':');
let kind = parts.next()?;
let preview_size = parts.next()?;
@@ -539,7 +540,10 @@ fn server_large_value_marker(value: &serde_json::Value) -> Option<ServerLargeVal
_ => return None,
};
let original_bytes = parts.next().and_then(|value| value.parse::<usize>().ok());
Some(ServerLargeValueMarkerValue { kind, preview_size: preview_size.parse::<usize>().ok()?.max(1), original_bytes })
let value_ref = parts.next().filter(|value| !value.is_empty());
Some(ServerLargeValueMarkerValue {
kind, preview_size: preview_size.parse::<usize>().ok()?.max(1), original_bytes, value_ref,
})
}
fn truncate_server_large_value_preview(
@@ -629,18 +633,23 @@ fn extract_server_large_value_markers(result: &mut db::QueryResult) -> Vec<db::L
}
for (row_index, row) in result.rows.iter_mut().enumerate() {
for marker in &markers {
let marker_value = row.get(marker.result_index).and_then(server_large_value_marker);
// Own the locator before mutating the preview in this row.
let value_ref = row.get(marker.result_index).and_then(server_large_value_marker)
.and_then(|value| value.value_ref).map(str::to_owned);
let marker_value = row.get(marker.result_index).and_then(server_large_value_marker)
.map(|value| (value.kind, value.preview_size, value.original_bytes));
let source_result_index = marker.result_index.saturating_sub(1);
if marker_value.is_some_and(|value| {
row.get_mut(source_result_index)
.is_some_and(|source| truncate_server_large_value_preview(source, value.kind, value.preview_size))
.is_some_and(|source| truncate_server_large_value_preview(source, value.0, value.1))
}) {
large_value_cells.push(db::LargeValueCell {
row_index,
column_index: marker.source_index,
original_bytes: marker_value
.and_then(|value| value.original_bytes)
.and_then(|value| value.2)
.unwrap_or(SERVER_LARGE_VALUE_UNKNOWN_BYTES),
value_ref,
});
}
}
@@ -3101,6 +3110,125 @@ fn parse_drop_database_target(sql: &str) -> Option<String> {
parts[0].as_ident().map(|ident| ident.value.clone())
}
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LargeValueRequest {
pub connection_id: String,
pub database: String,
pub value_ref: String,
#[serde(default)]
pub offset: u64,
#[serde(default)]
pub limit: u32,
pub execution_id: Option<String>,
pub client_session_id: Option<String>,
pub catalog: Option<String>,
pub txn_session_id: Option<String>,
}
/// Fetches only the original captured locator; never retries by selecting the current row.
pub async fn request_large_value(
state: &AppState,
request: LargeValueRequest,
release: bool,
) -> Result<serde_json::Value, String> {
if request.value_ref.len() > 128 || (!release && (request.limit == 0 || request.limit > 4096)) {
return Err("Invalid LOB chunk request".to_string());
}
let registered = request.execution_id.as_ref().map(|id| state.running_queries.register_task(
id.clone(),
crate::query_cancel::RunningTaskMetadata::query(
request.connection_id.clone(), request.database.clone(), request.client_session_id.clone(),
),
));
let cancel_token = registered.as_ref().map(|task| task.token());
request_large_value_with_cancel(state, request, release, cancel_token).await
}
async fn request_large_value_with_cancel(
state: &AppState,
request: LargeValueRequest,
release: bool,
cancel_token: Option<CancellationToken>,
) -> Result<serde_json::Value, String> {
let method = if release { "release_large_value" } else { "read_large_value_chunk" };
let params = serde_json::json!({ "valueRef": request.value_ref, "offset": request.offset, "limit": request.limit });
if let Some(txn_id) = request.txn_session_id.as_deref() {
let connection = {
let mut sessions = state.transaction_sessions.write().await;
let session = sessions.get_mut(txn_id).ok_or(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR)?;
if session.connection_id != request.connection_id || session.busy
|| session.last_activity.elapsed().as_secs() >= MANUAL_TRANSACTION_IDLE_TIMEOUT_SECS {
return Err("LOB transaction context expired or busy".to_string());
}
session.last_activity = std::time::Instant::now();
Arc::clone(&session.connection)
};
let connection = connection.lock().await;
let TxnConnection::Agent { client, .. } = &*connection else {
return Err("LOB snapshot chunks require an agent transaction".to_string());
};
let result = client.lock().await.call_with_timeout_and_cancel(
method, params, Some(std::time::Duration::from_secs(30)), cancel_token,
).await;
if !state.transaction_sessions.read().await.contains_key(txn_id) {
return Err("LOB transaction context expired".to_string());
}
return result;
}
let pool_database = query_pool_database(&request.database, request.catalog.as_deref());
let pool_key = state.get_or_create_pool_for_session(
&request.connection_id, pool_database, request.client_session_id.as_deref(),
).await?;
let _activity = state.pool_activity_touch(pool_key.as_str());
let client = {
let handle = state.pool_handle(&pool_key).await;
let Some(PoolKind::Agent(client)) = handle.as_ref() else {
return Err("LOB snapshot chunks require an agent connection".to_string());
};
client.clone()
};
let mut client = client.lock().await;
client.call_with_timeout_and_cancel(
method, params,
Some(std::time::Duration::from_secs(30)), cancel_token,
).await
}
pub async fn write_large_value_snapshot(
state: &AppState,
mut request: LargeValueRequest,
output: &mut std::fs::File,
) -> Result<u64, String> {
use std::io::Write;
let registered = request.execution_id.as_ref().map(|id| state.running_queries.register_task(
id.clone(), crate::query_cancel::RunningTaskMetadata::query(
request.connection_id.clone(), request.database.clone(), request.client_session_id.clone(),
),
));
let cancel = registered.as_ref().map(|task| task.token());
request.offset = 0;
request.limit = 4096;
let mut written = 0;
loop {
if is_canceled(&cancel) { return Err(canceled_error()); }
let chunk = request_large_value_with_cancel(state, request.clone(), false, cancel.clone()).await?;
if chunk.get("status").and_then(|value| value.as_str()) != Some("ok") {
return Err("LOB snapshot expired; execute the query again".to_string());
}
let next = chunk.get("next_offset").and_then(|value| value.as_u64()).ok_or("Invalid LOB offset")?;
let eof = chunk.get("eof").and_then(|value| value.as_bool()).ok_or("Invalid LOB EOF")?;
let data = chunk.get("data").and_then(|value| value.as_str()).ok_or("Invalid LOB data")?;
if next < request.offset || (!eof && next == request.offset) { return Err("Invalid LOB offset".to_string()); }
output.write_all(data.as_bytes()).map_err(|error| error.to_string())?;
written += data.len() as u64;
if eof { break; }
request.offset = next;
}
output.flush().map_err(|error| error.to_string())?;
Ok(written)
}
pub async fn close_query_session(
state: &AppState,
connection_id: &str,
@@ -9934,7 +10062,7 @@ for line in sys.stdin:
let result = ExecuteMultiResult::success_with_index_and_large_values(
empty_query_result(1),
0,
vec![db::LargeValueCell { row_index: 2, column_index: 3, original_bytes: 65_536 }],
vec![db::LargeValueCell { row_index: 2, column_index: 3, original_bytes: 65_536, value_ref: None }],
false,
);
@@ -9948,28 +10076,28 @@ for line in sys.stdin:
#[test]
fn large_value_cell_merge_replaces_driver_entries_and_appends_server_entries() {
let driver_cells = vec![
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 10 },
db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20 },
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 10, value_ref: None },
db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20, value_ref: None },
];
let server_cells = vec![
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 100 },
db::LargeValueCell { row_index: 5, column_index: 6, original_bytes: 200 },
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 100, value_ref: None },
db::LargeValueCell { row_index: 5, column_index: 6, original_bytes: 200, value_ref: None },
];
assert_eq!(
merge_large_value_cells(driver_cells, server_cells),
vec![
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 100 },
db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20 },
db::LargeValueCell { row_index: 5, column_index: 6, original_bytes: 200 },
db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 100, value_ref: None },
db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20, value_ref: None },
db::LargeValueCell { row_index: 5, column_index: 6, original_bytes: 200, value_ref: None },
]
);
}
#[test]
fn large_value_cell_merge_preserves_single_source_inputs() {
let driver_cell = db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 10 };
let server_cell = db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20 };
let driver_cell = db::LargeValueCell { row_index: 1, column_index: 2, original_bytes: 10, value_ref: None };
let server_cell = db::LargeValueCell { row_index: 3, column_index: 4, original_bytes: 20, value_ref: None };
assert_eq!(merge_large_value_cells(vec![driver_cell.clone()], Vec::new()), vec![driver_cell]);
assert_eq!(merge_large_value_cells(Vec::new(), vec![server_cell.clone()]), vec![server_cell]);
@@ -9979,10 +10107,10 @@ for line in sys.stdin:
fn large_value_cell_merge_handles_large_disjoint_inputs() {
const CELL_COUNT: usize = 100_000;
let driver_cells = (0..CELL_COUNT)
.map(|row_index| db::LargeValueCell { row_index, column_index: 0, original_bytes: 10 })
.map(|row_index| db::LargeValueCell { row_index, column_index: 0, original_bytes: 10, value_ref: None })
.collect();
let server_cells = (0..CELL_COUNT)
.map(|row_index| db::LargeValueCell { row_index, column_index: 1, original_bytes: 20 })
.map(|row_index| db::LargeValueCell { row_index, column_index: 1, original_bytes: 20, value_ref: None })
.collect();
let merged = merge_large_value_cells(driver_cells, server_cells);
@@ -10307,6 +10435,33 @@ for line in sys.stdin:
));
}
#[test]
fn original_lob_ref_survives_production_result_normalization_without_polluting_preview_values() {
let mut result = empty_query_result(0);
result.columns = vec!["Payload".to_string(), "__DBX_LARGE_VALUE_BYTES_C_0".to_string()];
result.column_types = vec!["CLOB".to_string(), "VARCHAR".to_string()];
result.rows = vec![
vec![serde_json::json!("原😀值"), serde_json::json!("D:1:unknown:original-ref")],
vec![serde_json::Value::Null, serde_json::Value::Null],
vec![serde_json::json!(""), serde_json::Value::Null],
];
let normalized = ExecuteMultiResult::success_with_optional_server_large_values(result, true);
assert_eq!(normalized.result.columns, vec!["Payload"]);
assert_eq!(normalized.result.rows, vec![vec![serde_json::json!("原😀值")], vec![serde_json::Value::Null], vec![serde_json::json!("")]]);
assert_eq!(normalized.large_value_cells, vec![db::LargeValueCell {
row_index: 0, column_index: 0, original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES,
value_ref: Some("original-ref".to_string()),
}]);
let wire = serde_json::to_value(normalized).unwrap();
assert_eq!(wire["large_value_cells"][0]["value_ref"], "original-ref");
assert_eq!(wire["rows"][0][0], "原😀值");
let legacy: db::LargeValueCell = serde_json::from_value(serde_json::json!({
"row_index": 0, "column_index": 1, "original_bytes": 128,
})).unwrap();
assert_eq!(legacy.value_ref, None);
assert!(serde_json::to_value(legacy).unwrap().get("value_ref").is_none());
}
#[test]
fn external_driver_preview_fallback_rewrites_generated_postgres_projections() {
let sql = concat!(
@@ -10464,8 +10619,7 @@ for line in sys.stdin:
vec![db::LargeValueCell {
row_index: 0,
column_index: 1,
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES,
}]
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES, value_ref: None }]
);
let options = QueryExecutionOptions { max_rows: Some(100), table_data_preview: true, ..Default::default() };
@@ -12199,8 +12353,7 @@ for line in sys.stdin:
vec![db::LargeValueCell {
row_index: 0,
column_index: 1,
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES,
}]
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES, value_ref: None }]
);
}
@@ -12266,8 +12419,8 @@ for line in sys.stdin:
assert_eq!(
cells,
vec![
db::LargeValueCell { row_index: 0, column_index: 1, original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES },
db::LargeValueCell { row_index: 0, column_index: 2, original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES },
db::LargeValueCell { row_index: 0, column_index: 1, original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES, value_ref: None },
db::LargeValueCell { row_index: 0, column_index: 2, original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES, value_ref: None },
]
);
}
@@ -12324,7 +12477,7 @@ for line in sys.stdin:
cells,
[3, 7, 11]
.into_iter()
.map(|row_index| db::LargeValueCell { row_index, column_index: 0, original_bytes: 5 })
.map(|row_index| db::LargeValueCell { row_index, column_index: 0, original_bytes: 5, value_ref: None })
.collect::<Vec<_>>()
);
}
@@ -12427,7 +12580,7 @@ for line in sys.stdin:
.enumerate()
.filter_map(|(row_index, size)| size
.filter(|size| *size > preview_size)
.map(|original_bytes| { db::LargeValueCell { row_index, column_index: 1, original_bytes } }))
.map(|original_bytes| { db::LargeValueCell { row_index, column_index: 1, original_bytes, value_ref: None } }))
.collect::<Vec<_>>()
);
}
@@ -12471,8 +12624,7 @@ for line in sys.stdin:
vec![db::LargeValueCell {
row_index: 0,
column_index: 1,
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES,
}]
original_bytes: SERVER_LARGE_VALUE_UNKNOWN_BYTES, value_ref: None }]
);
}
+4 -4
View File
@@ -1288,7 +1288,7 @@ fn mysql_row_to_json_with_srids_and_previews(
if let Some((preview, original_bytes)) =
preview_bytes.and_then(|limit| mysql_bounded_value_preview(row, index, limit, protected_indexes))
{
large_value_cells.push(LargeValueCell { row_index, column_index: index, original_bytes });
large_value_cells.push(LargeValueCell { row_index, column_index: index, original_bytes, value_ref: None });
return preview;
}
mysql_value_to_json_with_effective_type(row, index, effective_types.get(index).map(String::as_str))
@@ -9109,7 +9109,7 @@ mod tests {
assert_eq!(values[0], serde_json::json!("k".repeat(2048)));
assert_eq!(values[1].as_str().map(str::len), Some(1027));
assert!(values[1].as_str().is_some_and(|value| value.ends_with("...")));
assert_eq!(cells, vec![LargeValueCell { row_index: 7, column_index: 1, original_bytes: 32 * 1024 }]);
assert_eq!(cells, vec![LargeValueCell { row_index: 7, column_index: 1, original_bytes: 32 * 1024, value_ref: None }]);
}
#[test]
@@ -9134,7 +9134,7 @@ mod tests {
mysql_row_to_json_with_srids_and_previews(&row, &mut spatial_columns, 0, Some(512), &protected, &[]);
assert_eq!(values[0], serde_json::json!("k".repeat(2048)));
assert_eq!(cells, vec![LargeValueCell { row_index: 0, column_index: 1, original_bytes: 4096 }]);
assert_eq!(cells, vec![LargeValueCell { row_index: 0, column_index: 1, original_bytes: 4096, value_ref: None }]);
}
#[test]
@@ -9174,7 +9174,7 @@ mod tests {
assert!(values[1].as_str().is_some_and(|value| value.ends_with("...")));
assert_eq!(values[2].as_str().map(str::len), Some(2 + 420 * 2));
assert_eq!(values[3], serde_json::json!("B:419:25143"));
assert_eq!(cells, vec![LargeValueCell { row_index: 0, column_index: 1, original_bytes: 10_000 }]);
assert_eq!(cells, vec![LargeValueCell { row_index: 0, column_index: 1, original_bytes: 10_000, value_ref: None }]);
}
#[test]
+3
View File
@@ -545,6 +545,9 @@ pub struct LargeValueCell {
pub row_index: usize,
pub column_index: usize,
pub original_bytes: usize,
/// Opaque original-result locator. Absent for legacy preview/re-query drivers.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub value_ref: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
+4
View File
@@ -913,6 +913,10 @@ async fn serve() -> Result<(), String> {
.route("/query/cancel", post(routes::query::cancel_query))
.route("/query/cancel-conditional-update", post(routes::query::cancel_conditional_update))
.route("/query/close-session", post(routes::query::close_query_session))
.route("/query/large-value/chunk", post(routes::query::read_large_value_chunk))
.route("/query/large-value/release", post(routes::query::release_large_value))
.route("/query/large-value/download", post(routes::query::prepare_large_value_download))
.route("/query/large-value/download/{id}", get(routes::query::download_large_value))
.route("/query/close-client-session", post(routes::query::close_client_connection_session))
.route("/export/query-result-json", post(routes::text_export::export_query_result_json))
.route("/export/query-result-markdown", post(routes::text_export::export_query_result_markdown))
+55
View File
@@ -677,6 +677,61 @@ pub async fn cancel_conditional_update(
Json(state.app.running_queries.cancel_and_wait(&req.execution_id, CONDITIONAL_UPDATE_CANCEL_WAIT).await)
}
pub async fn read_large_value_chunk(
State(state): State<Arc<WebState>>,
Json(req): Json<dbx_core::query::LargeValueRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
dbx_core::query::request_large_value(&state.app, req, false).await
.map(Json).map_err(AppError::from)
}
pub async fn release_large_value(
State(state): State<Arc<WebState>>,
Json(req): Json<dbx_core::query::LargeValueRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
dbx_core::query::request_large_value(&state.app, req, true).await
.map(Json).map_err(AppError::from)
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LargeValueDownloadRequest {
pub request: dbx_core::query::LargeValueRequest,
pub file_name: String,
}
pub async fn prepare_large_value_download(
State(state): State<Arc<WebState>>,
Json(req): Json<LargeValueDownloadRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
let mut temporary = tempfile::NamedTempFile::new().map_err(|error| AppError::from(error.to_string()))?;
dbx_core::query::write_large_value_snapshot(&state.app, req.request, temporary.as_file_mut())
.await.map_err(AppError::from)?;
let (_, path) = temporary.keep().map_err(|error| AppError::from(error.to_string()))?;
let download_id = uuid::Uuid::new_v4().to_string();
state.export_files.write().await.insert(download_id.clone(), crate::state::WebExportFile {
file_path: path.to_string_lossy().into_owned(), download_filename: req.file_name, format: "lob".to_string(),
});
let cleanup_state = state.clone();
let cleanup_id = download_id.clone();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(300)).await;
if let Some(file) = cleanup_state.export_files.write().await.remove(&cleanup_id) {
let _ = tokio::fs::remove_file(file.file_path).await;
}
});
Ok(Json(serde_json::json!({ "downloadId": download_id })))
}
pub async fn download_large_value(
State(state): State<Arc<WebState>>,
axum::extract::Path(id): axum::extract::Path<String>,
) -> Result<axum::response::Response, AppError> {
let file = state.export_files.write().await.remove(&id)
.ok_or_else(|| AppError::from("LOB download expired".to_string()))?;
super::mongodb_import_export::export_file_response(file).await
}
pub async fn close_query_session(
State(state): State<Arc<WebState>>,
Json(req): Json<CloseSessionRequest>,
+30
View File
@@ -334,6 +334,36 @@ pub async fn close_query_session(
.await
}
#[tauri::command]
pub async fn read_large_value_chunk(
state: State<'_, Arc<AppState>>,
request: dbx_core::query::LargeValueRequest,
) -> Result<serde_json::Value, String> {
dbx_core::query::request_large_value(&state, request, false).await
}
#[tauri::command]
pub async fn release_large_value(
state: State<'_, Arc<AppState>>,
request: dbx_core::query::LargeValueRequest,
) -> Result<serde_json::Value, String> {
dbx_core::query::request_large_value(&state, request, true).await
}
#[tauri::command]
pub async fn download_large_value(
state: State<'_, Arc<AppState>>,
request: dbx_core::query::LargeValueRequest,
file_path: String,
) -> Result<u64, String> {
let path = std::path::Path::new(&file_path);
let parent = path.parent().filter(|parent| !parent.as_os_str().is_empty()).unwrap_or(std::path::Path::new("."));
let mut temporary = tempfile::NamedTempFile::new_in(parent).map_err(|error| error.to_string())?;
let written = dbx_core::query::write_large_value_snapshot(&state, request, temporary.as_file_mut()).await?;
temporary.persist(path).map_err(|error| error.to_string())?;
Ok(written)
}
#[tauri::command]
pub async fn close_client_connection_session(
state: State<'_, Arc<AppState>>,
+3
View File
@@ -2076,6 +2076,9 @@ pub fn run() {
commands::query::cancel_query,
commands::query::cancel_conditional_update,
commands::query::close_query_session,
commands::query::read_large_value_chunk,
commands::query::release_large_value,
commands::query::download_large_value,
commands::query::close_client_connection_session,
commands::query::execute_batch,
commands::query::execute_script,