From 989dfb79bb723081eb7884679dfd4d167e6f5591 Mon Sep 17 00:00:00 2001 From: efeire <15682810+efeire@users.noreply.github.com> Date: Fri, 9 Oct 2026 18:04:45 +0800 Subject: [PATCH] feat(oceanbase): retain original CLOB snapshots for bounded value reads --- .../java/com/dbx/agent/AbstractJdbcAgent.java | 10 +- .../java/com/dbx/agent/AgentProtocol.java | 4 + .../java/com/dbx/agent/DatabaseAgent.java | 8 + .../main/java/com/dbx/agent/JdbcExecutor.java | 14 ++ .../java/com/dbx/agent/JsonRpcServer.java | 13 ++ .../com/dbx/agent/JsonRpcLargeValueTest.java | 49 ++++ .../oceanbaseoracle/OceanBaseLobValues.java | 214 ++++++++++++++++++ .../oceanbaseoracle/OceanBaseOracleAgent.java | 117 ++++++++++ .../OceanBaseLobSnapshotIntegrationTest.java | 67 ++++++ .../OceanBaseLobValuesTest.java | 140 ++++++++++++ apps/desktop/src/components/grid/DataGrid.vue | 18 ++ .../src/composables/useDataGridExport.ts | 12 +- .../src/composables/useDataGridLargeValues.ts | 62 ++++- apps/desktop/src/i18n/locales/en.ts | 2 + apps/desktop/src/i18n/locales/zh-CN.ts | 2 + apps/desktop/src/i18n/locales/zh-TW.ts | 2 + .../dataGrid/largeValueSnapshot.spec.ts | 64 ++++++ apps/desktop/src/lib/backend/api.ts | 3 + apps/desktop/src/lib/backend/http.ts | 37 +++ apps/desktop/src/lib/backend/tauri.ts | 12 + .../src/lib/dataGrid/largeValueSnapshot.ts | 61 +++++ apps/desktop/src/stores/queryStore.ts | 48 +++- apps/desktop/src/types/database.ts | 4 +- crates/dbx-core/src/query/mod.rs | 208 ++++++++++++++--- crates/dbx-driver-mysql/src/mysql.rs | 8 +- crates/dbx-types/src/types.rs | 3 + crates/dbx-web/src/main.rs | 4 + crates/dbx-web/src/routes/query.rs | 55 +++++ src-tauri/src/commands/query.rs | 30 +++ src-tauri/src/lib.rs | 3 + 30 files changed, 1232 insertions(+), 42 deletions(-) create mode 100644 agents/common/src/test/java/com/dbx/agent/JsonRpcLargeValueTest.java create mode 100644 agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValues.java create mode 100644 agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobSnapshotIntegrationTest.java create mode 100644 agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValuesTest.java create mode 100644 apps/desktop/src/lib/__tests__/dataGrid/largeValueSnapshot.spec.ts create mode 100644 apps/desktop/src/lib/dataGrid/largeValueSnapshot.ts diff --git a/agents/common/src/main/java/com/dbx/agent/AbstractJdbcAgent.java b/agents/common/src/main/java/com/dbx/agent/AbstractJdbcAgent.java index 528eb48044..29895eb153 100644 --- a/agents/common/src/main/java/com/dbx/agent/AbstractJdbcAgent.java +++ b/agents/common/src/main/java/com/dbx/agent/AbstractJdbcAgent.java @@ -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; diff --git a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java index 93cd3d388f..b23c679d43 100644 --- a/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java +++ b/agents/common/src/main/java/com/dbx/agent/AgentProtocol.java @@ -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, diff --git a/agents/common/src/main/java/com/dbx/agent/DatabaseAgent.java b/agents/common/src/main/java/com/dbx/agent/DatabaseAgent.java index d952b2f3d2..9557c5a648 100644 --- a/agents/common/src/main/java/com/dbx/agent/DatabaseAgent.java +++ b/agents/common/src/main/java/com/dbx/agent/DatabaseAgent.java @@ -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) { diff --git a/agents/common/src/main/java/com/dbx/agent/JdbcExecutor.java b/agents/common/src/main/java/com/dbx/agent/JdbcExecutor.java index ea6193fff8..ed9cb79332 100644 --- a/agents/common/src/main/java/com/dbx/agent/JdbcExecutor.java +++ b/agents/common/src/main/java/com/dbx/agent/JdbcExecutor.java @@ -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 withActiveStatement(Statement statement, SqlOperation operation) throws SQLException { + activeStatements.add(statement); + try { + return operation.run(); + } finally { + activeStatements.remove(statement); + } + } + + @FunctionalInterface + public interface SqlOperation { + T run() throws SQLException; + } + public interface ColumnAwareResultValueReader extends ResultValueReader { Object read(ResultSet rs, int index, int sqlType, String columnTypeName) throws SQLException; diff --git a/agents/common/src/main/java/com/dbx/agent/JsonRpcServer.java b/agents/common/src/main/java/com/dbx/agent/JsonRpcServer.java index 7a232cb223..0d5a7cb1e3 100644 --- a/agents/common/src/main/java/com/dbx/agent/JsonRpcServer.java +++ b/agents/common/src/main/java/com/dbx/agent/JsonRpcServer.java @@ -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>() {}.getType(); List 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>() {}.getType(); List statements = gson.fromJson(params.get("statements"), statementsType); return agent.executeBatch(statements, stringOrNull(params, "schema")); diff --git a/agents/common/src/test/java/com/dbx/agent/JsonRpcLargeValueTest.java b/agents/common/src/test/java/com/dbx/agent/JsonRpcLargeValueTest.java new file mode 100644 index 0000000000..7af2b2b12c --- /dev/null +++ b/agents/common/src/test/java/com/dbx/agent/JsonRpcLargeValueTest.java @@ -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 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 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); + } +} diff --git a/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValues.java b/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValues.java new file mode 100644 index 0000000000..4df95f295a --- /dev/null +++ b/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValues.java @@ -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 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 columns, List types, List> 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 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 columns, List types, List> rows) { + List expandedColumns = new ArrayList<>(); + List expandedTypes = new ArrayList<>(); + boolean[] marked = new boolean[columns.size()]; + for (List 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> expandedRows = new ArrayList<>(); + for (List row : rows) { + List 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); + } +} diff --git a/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleAgent.java b/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleAgent.java index 3ff0c511ac..497d424aed 100644 --- a/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleAgent.java +++ b/agents/drivers/oceanbase-oracle/src/main/java/com/dbx/agent/oceanbaseoracle/OceanBaseOracleAgent.java @@ -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 remapLobSpatialColumns(List spatial, List columns) { + List retained = new ArrayList<>(); + for (int index = 0; index < columns.size(); index++) { + if (!columns.get(index).startsWith(OceanBaseLobValues.MARKER_PREFIX)) retained.add(index); + } + List remapped = new ArrayList<>(); + for (SpatialColumn column : spatial) remapped.add(new SpatialColumn(retained.get(column.getColumn_index()), column.getSrid())); + return remapped; + } + + private static List> remapLobSpatialValues(List> spatial, List columns) { + List> remapped = new ArrayList<>(); + for (List row : spatial) { + List 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); diff --git a/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobSnapshotIntegrationTest.java b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobSnapshotIntegrationTest.java new file mode 100644 index 0000000000..0949702c78 --- /dev/null +++ b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobSnapshotIntegrationTest.java @@ -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"); } + } + } + } +} diff --git a/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValuesTest.java b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValuesTest.java new file mode 100644 index 0000000000..abb279bdf6 --- /dev/null +++ b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobValuesTest.java @@ -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 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 proxy(Class type, Handler handler) { + return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, + (object, method, args) -> handler.call(method.getName(), args))); + } +} diff --git a/apps/desktop/src/components/grid/DataGrid.vue b/apps/desktop/src/components/grid/DataGrid.vue index e187b9fe35..c3cddd2296 100644 --- a/apps/desktop/src/components/grid/DataGrid.vue +++ b/apps/desktop/src/components/grid/DataGrid.vue @@ -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"), diff --git a/apps/desktop/src/composables/useDataGridExport.ts b/apps/desktop/src/composables/useDataGridExport.ts index 4838cbb106..abc9068fb2 100644 --- a/apps/desktop/src/composables/useDataGridExport.ts +++ b/apps/desktop/src/composables/useDataGridExport.ts @@ -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); diff --git a/apps/desktop/src/composables/useDataGridLargeValues.ts b/apps/desktop/src/composables/useDataGridLargeValues.ts index 931dcbd680..1b1771c3e9 100644 --- a/apps/desktop/src/composables/useDataGridLargeValues.ts +++ b/apps/desktop/src/composables/useDataGridLargeValues.ts @@ -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(); + const snapshotExecutionIds = new Set(); + 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 { + 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 { + 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(); @@ -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(); + 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, diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 260b4bf1f7..d4315a7280 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -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: { diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 91e247434e..91c39a9900 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -2797,6 +2797,8 @@ export default withEnglishFallback({ binaryImportFailed: "导入文件失败：{message}", binaryImportTooLarge: "文件过大（{size}，上限 {limit}）。不支持将大文件导入到单个单元格。", downloadBinaryValue: "下载值", + downloadSnapshotValue: "下载完整结果快照", + largeValueSnapshotPreview: "{value}（结果快照预览）", downloadSaved: "已保存到 {path}", downloadStarted: "已交给浏览器下载：{fileName}", binaryDownload: { diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index c673433820..668f49024b 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -2572,6 +2572,8 @@ export default withEnglishFallback({ binaryImportFailed: "匯入檔案失敗：{message}", binaryImportTooLarge: "檔案過大（{size}，上限 {limit}）。不支援將大檔案匯入到單一儲存格。", downloadBinaryValue: "下載值", + downloadSnapshotValue: "下載完整結果快照", + largeValueSnapshotPreview: "{value}（結果快照預覽）", downloadSaved: "已儲存到 {path}", downloadStarted: "已交給瀏覽器下載：{fileName}", binaryDownload: { diff --git a/apps/desktop/src/lib/__tests__/dataGrid/largeValueSnapshot.spec.ts b/apps/desktop/src/lib/__tests__/dataGrid/largeValueSnapshot.spec.ts new file mode 100644 index 0000000000..379f324f17 --- /dev/null +++ b/apps/desktop/src/lib/__tests__/dataGrid/largeValueSnapshot.spec.ts @@ -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", + })); + }); +}); diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts index eb6428f451..fcab597f7f 100644 --- a/apps/desktop/src/lib/backend/api.ts +++ b/apps/desktop/src/lib/backend/api.ts @@ -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"); diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index 6973a96042..501b712c63 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -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 { + return post("/api/query/large-value/chunk", request); +} + +export async function releaseLargeValue(request: LargeValueRequest): Promise { + return post("/api/query/large-value/release", request); +} + +export async function downloadLargeValue(request: LargeValueRequest, filePath: string): Promise { + 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 { return post("/api/query/close-session", { connectionId, diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index 208ea2d65f..3476cc0df2 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -2024,6 +2024,18 @@ export async function cancelConditionalUpdate(executionId: string): Promise { + return invoke("read_large_value_chunk", { request }); +} + +export async function releaseLargeValue(request: import("./http").LargeValueRequest): Promise { + return invoke("release_large_value", { request }); +} + +export async function downloadLargeValue(request: import("./http").LargeValueRequest, filePath: string): Promise { + await invoke("download_large_value", { request, filePath }); +} + export async function closeQuerySession(connectionId: string, database: string, sessionId: string, clientSessionId?: string, catalog?: string): Promise { return invoke("close_query_session", { connectionId, diff --git a/apps/desktop/src/lib/dataGrid/largeValueSnapshot.ts b/apps/desktop/src/lib/dataGrid/largeValueSnapshot.ts new file mode 100644 index 0000000000..5b0b3bb295 --- /dev/null +++ b/apps/desktop/src/lib/dataGrid/largeValueSnapshot.ts @@ -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, +): Promise { + 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 { + 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 { + 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; +} diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index 81811831b6..de81bb0576 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -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) { + const released = new Set(); + 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[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. diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts index 00a731a35a..4e23b265da 100644 --- a/apps/desktop/src/types/database.ts +++ b/apps/desktop/src/types/database.ts @@ -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, diff --git a/crates/dbx-core/src/query/mod.rs b/crates/dbx-core/src/query/mod.rs index 66d758b492..189c0189b4 100644 --- a/crates/dbx-core/src/query/mod.rs +++ b/crates/dbx-core/src/query/mod.rs @@ -482,10 +482,11 @@ struct ServerLargeValueMarker { } #[derive(Clone, Copy)] -struct ServerLargeValueMarkerValue { +struct ServerLargeValueMarkerValue<'a> { kind: ServerLargeValuePreviewKind, preview_size: usize, original_bytes: Option, + 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 { +fn server_large_value_marker(value: &serde_json::Value) -> Option> { 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 return None, }; let original_bytes = parts.next().and_then(|value| value.parse::().ok()); - Some(ServerLargeValueMarkerValue { kind, preview_size: preview_size.parse::().ok()?.max(1), original_bytes }) + let value_ref = parts.next().filter(|value| !value.is_empty()); + Some(ServerLargeValueMarkerValue { + kind, preview_size: preview_size.parse::().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 Option { 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, + pub client_session_id: Option, + pub catalog: Option, + pub txn_session_id: Option, +} + +/// 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 { + 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, +) -> Result { + 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 { + 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::>() ); } @@ -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::>() ); } @@ -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 }] ); } diff --git a/crates/dbx-driver-mysql/src/mysql.rs b/crates/dbx-driver-mysql/src/mysql.rs index 998f463147..7e1e993c33 100644 --- a/crates/dbx-driver-mysql/src/mysql.rs +++ b/crates/dbx-driver-mysql/src/mysql.rs @@ -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] diff --git a/crates/dbx-types/src/types.rs b/crates/dbx-types/src/types.rs index 77eed42909..c324e4f99e 100644 --- a/crates/dbx-types/src/types.rs +++ b/crates/dbx-types/src/types.rs @@ -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, } #[derive(Debug, Clone, Deserialize)] diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 8e3d15c410..bba528a353 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -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)) diff --git a/crates/dbx-web/src/routes/query.rs b/crates/dbx-web/src/routes/query.rs index e9d92c5556..d5bef8581d 100644 --- a/crates/dbx-web/src/routes/query.rs +++ b/crates/dbx-web/src/routes/query.rs @@ -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>, + Json(req): Json, +) -> Result, 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>, + Json(req): Json, +) -> Result, 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>, + Json(req): Json, +) -> Result, 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>, + axum::extract::Path(id): axum::extract::Path, +) -> Result { + 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>, Json(req): Json, diff --git a/src-tauri/src/commands/query.rs b/src-tauri/src/commands/query.rs index 4161382041..45116e50a8 100644 --- a/src-tauri/src/commands/query.rs +++ b/src-tauri/src/commands/query.rs @@ -334,6 +334,36 @@ pub async fn close_query_session( .await } +#[tauri::command] +pub async fn read_large_value_chunk( + state: State<'_, Arc>, + request: dbx_core::query::LargeValueRequest, +) -> Result { + dbx_core::query::request_large_value(&state, request, false).await +} + +#[tauri::command] +pub async fn release_large_value( + state: State<'_, Arc>, + request: dbx_core::query::LargeValueRequest, +) -> Result { + dbx_core::query::request_large_value(&state, request, true).await +} + +#[tauri::command] +pub async fn download_large_value( + state: State<'_, Arc>, + request: dbx_core::query::LargeValueRequest, + file_path: String, +) -> Result { + 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>, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 24e84eef3d..b6a3c19401 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -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,