From ab81d1fd8bc7ba84d2e1e16729cb6d43ff3b4be8 Mon Sep 17 00:00:00 2001 From: efeire <15682810+efeire@users.noreply.github.com> Date: Sat, 10 Oct 2026 02:38:15 +0800 Subject: [PATCH] fix(oceanbase): preserve pooled LOB lifecycle and pending grid changes --- .../src/main/resources/agent-protocol-v1.json | 2 + .../src/main/resources/agent-protocol-v2.json | 2 + .../oceanbaseoracle/OceanBaseLobValues.java | 11 +- .../OceanBaseLobStatementTest.java | 104 ++++++++++++++++++ .../__tests__/useDataGridLargeValues.spec.ts | 61 +++++++++- .../src/composables/useDataGridLargeValues.ts | 6 +- apps/desktop/src/stores/queryStore.ts | 4 +- crates/dbx-core/assets/agent-protocol-v1.json | 2 + .../assets/agent-protocol-v2.json | 2 + crates/dbx-driver-agent/src/agent_driver.rs | 6 + src-tauri/Cargo.toml | 2 +- 11 files changed, 191 insertions(+), 11 deletions(-) create mode 100644 agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobStatementTest.java diff --git a/agents/common/src/main/resources/agent-protocol-v1.json b/agents/common/src/main/resources/agent-protocol-v1.json index b363666a9e..af91085bbf 100644 --- a/agents/common/src/main/resources/agent-protocol-v1.json +++ b/agents/common/src/main/resources/agent-protocol-v1.json @@ -32,6 +32,8 @@ "execute_query_page", "fetch_query_page", "close_query_session", + "read_large_value_chunk", + "release_large_value", "start_table_read", "fetch_table_read_page", "close_table_read_session", diff --git a/agents/common/src/main/resources/agent-protocol-v2.json b/agents/common/src/main/resources/agent-protocol-v2.json index 8cd5f4e230..b0f3ae3046 100644 --- a/agents/common/src/main/resources/agent-protocol-v2.json +++ b/agents/common/src/main/resources/agent-protocol-v2.json @@ -36,6 +36,8 @@ "execute_query_page", "fetch_query_page", "close_query_session", + "read_large_value_chunk", + "release_large_value", "start_table_read", "fetch_table_read_page", "close_table_read_session", 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 index 15ab8fc56e..e1ee65c607 100644 --- 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 @@ -146,8 +146,15 @@ final class OceanBaseLobValues { 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(); + OceanBaseStatement vendor; + try { + vendor = call instanceof OceanBaseStatement ? (OceanBaseStatement) call : call.unwrap(OceanBaseStatement.class); + } catch (SQLException error) { + throw new SQLException("Unsupported OceanBase LOB statement", error); + } + if (vendor == null) throw new SQLException("Unsupported OceanBase LOB statement"); + // Only the vendor flag bypasses the pool proxy; lifecycle and cancellation stay on call. + vendor.setInternal(); call.setClob(1, locator); call.setInt(2, limit); call.setLong(3, offset + 1); diff --git a/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobStatementTest.java b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobStatementTest.java new file mode 100644 index 0000000000..c4dbf23af7 --- /dev/null +++ b/agents/drivers/oceanbase-oracle/src/test/java/com/dbx/agent/oceanbaseoracle/OceanBaseLobStatementTest.java @@ -0,0 +1,104 @@ +package com.dbx.agent.oceanbaseoracle; + +import com.dbx.agent.JdbcExecutor; +import com.oceanbase.jdbc.OceanBaseConnection; +import com.oceanbase.jdbc.OceanBaseStatement; +import com.oceanbase.jdbc.internal.protocol.Protocol; +import com.oceanbase.jdbc.util.Options; +import com.zaxxer.hikari.HikariDataSource; +import java.lang.reflect.Proxy; +import java.sql.CallableStatement; +import java.sql.Clob; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Types; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.ReentrantLock; +import javax.sql.DataSource; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +class OceanBaseLobStatementTest { + @Test + void pooledCallableUnwrapsVendorOnlyForInternalFlagAndKeepsLifecycleOnProxy() throws Exception { + { + List calls = new ArrayList<>(); + Options options = new Options(); + Protocol protocol = proxy(Protocol.class, (object, method, args) -> { + if (method.getName().equals("getOptions")) return options; + if (method.getName().equals("getLock")) return new ReentrantLock(); + return defaultValue(method.getReturnType()); + }); + OceanBaseStatement vendor = new OceanBaseStatement(new OceanBaseConnection(protocol), 1003, 1007, null); + Clob locator = proxy(Clob.class, (o, m, a) -> defaultValue(m.getReturnType())); + CallableStatement delegate = proxy(CallableStatement.class, (object, method, args) -> { + switch (method.getName()) { + case "unwrap": return ((Class) args[0]).cast(vendor); + case "isWrapperFor": return ((Class) args[0]).isInstance(vendor); + case "setClob": assertSame(locator, args[1]); calls.add(method.getName()); return null; + case "setInt": assertEquals(3, args[1]); return null; + case "setLong": assertEquals(6L, args[1]); return null; + case "setQueryTimeout": assertEquals(20, args[0]); return null; + case "registerOutParameter": assertEquals((int) args[0] == 2 ? Types.INTEGER : Types.VARCHAR, args[1]); return null; + case "execute": + assertTrue(vendor.isInternal()); + assertTrue(JdbcExecutor.current().hasActiveStatements()); + JdbcExecutor.current().cancelActiveStatements(); + calls.add("execute"); return false; + case "cancel", "close": calls.add(method.getName()); return null; + case "getInt": return 2; + case "getBytes": return "中😀".getBytes(java.nio.charset.StandardCharsets.UTF_8); + default: return defaultValue(method.getReturnType()); + } + }); + Connection physical = proxy(Connection.class, (object, method, args) -> { + if (method.getName().equals("prepareCall")) return delegate; + if (method.getName().equals("getAutoCommit") || method.getName().equals("isValid")) return true; + if (method.getName().equals("getTransactionIsolation")) return Connection.TRANSACTION_READ_COMMITTED; + return defaultValue(method.getReturnType()); + }); + DataSource source = proxy(DataSource.class, (object, method, args) -> method.getName().equals("getConnection") ? physical : defaultValue(method.getReturnType())); + try (HikariDataSource pool = new HikariDataSource()) { + pool.setDataSource(source); + pool.setMaximumPoolSize(1); + pool.setMinimumIdle(0); + try (Connection connection = pool.getConnection()) { + var chunk = OceanBaseLobValues.read(connection, locator, 5, 3); + assertEquals("中😀", chunk.data()); + assertEquals(7, chunk.next_offset()); + assertTrue(chunk.eof()); + assertEquals("text", chunk.value_kind()); + } + } + assertEquals(List.of("setClob", "cancel", "execute", "close"), calls); + assertFalse(JdbcExecutor.current().hasActiveStatements()); + } + } + + @Test + void nonOceanBaseCallableIsRejectedAndClosedBeforeExecuting() { + List calls = new ArrayList<>(); + CallableStatement call = proxy(CallableStatement.class, (object, method, args) -> { + if (method.getName().equals("unwrap")) throw new SQLException("Not a vendor wrapper"); + if (method.getName().equals("close") || method.getName().equals("execute")) calls.add(method.getName()); + return defaultValue(method.getReturnType()); + }); + Connection connection = proxy(Connection.class, (object, method, args) -> method.getName().equals("prepareCall") ? call : defaultValue(method.getReturnType())); + SQLException error = assertThrows(SQLException.class, () -> OceanBaseLobValues.read(connection, proxy(Clob.class, (o, m, a) -> null), 0, 1)); + assertEquals("Unsupported OceanBase LOB statement", error.getMessage()); + assertEquals(List.of("close"), calls); + assertFalse(JdbcExecutor.current().hasActiveStatements()); + } + + private static Object defaultValue(Class type) { + if (type == boolean.class) return false; + if (type == int.class) return 0; + if (type == long.class) return 0L; + return null; + } + + private static T proxy(Class type, java.lang.reflect.InvocationHandler handler) { + return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, handler)); + } +} diff --git a/apps/desktop/src/composables/__tests__/useDataGridLargeValues.spec.ts b/apps/desktop/src/composables/__tests__/useDataGridLargeValues.spec.ts index 95119352f1..66bcf3d5e9 100644 --- a/apps/desktop/src/composables/__tests__/useDataGridLargeValues.spec.ts +++ b/apps/desktop/src/composables/__tests__/useDataGridLargeValues.spec.ts @@ -1,9 +1,10 @@ // @vitest-environment happy-dom -import { computed, createApp, defineComponent, h, ref, type Ref } from "vue"; +import { computed, createApp, defineComponent, h, nextTick, ref, type Ref } from "vue"; import { afterEach, describe, expect, it, vi } from "vitest"; import { binaryCellDisplayText, binaryCellDownloadPayload } from "@/lib/dataGrid/binaryCellDownload"; import { useDataGridLargeValues } from "@/composables/useDataGridLargeValues"; +import { useDataGridEditor } from "@/composables/useDataGridEditor"; import type { DatabaseType, QueryResult } from "@/types/database"; const mocks = vi.hoisted(() => ({ @@ -13,6 +14,8 @@ const mocks = vi.hoisted(() => ({ executeMulti: vi.fn(), readLargeValueChunk: vi.fn(), downloadLargeValue: vi.fn(), + prepareDataGridSave: vi.fn(), + executeBatch: vi.fn(), })); vi.mock("@/lib/backend/api", () => ({ @@ -21,9 +24,15 @@ vi.mock("@/lib/backend/api", () => ({ executeMulti: mocks.executeMulti, readLargeValueChunk: mocks.readLargeValueChunk, downloadLargeValue: mocks.downloadLargeValue, + prepareDataGridSave: mocks.prepareDataGridSave, + executeBatch: mocks.executeBatch, + connectionWriteUnlockState: vi.fn().mockResolvedValue(0), })); vi.mock("@/lib/table/tableSelectSql", () => ({ buildTableSelectSql: mocks.buildTableSelectSql })); +vi.mock("@/stores/connectionStore", () => ({ useConnectionStore: () => ({ getConfig: () => undefined }) })); +vi.mock("@/stores/historyStore", () => ({ useHistoryStore: () => ({ add: vi.fn() }) })); +vi.mock("@/stores/productionSafetyStore", () => ({ useProductionSafetyStore: () => ({}) })); const disposers: Array<() => void> = []; @@ -32,11 +41,12 @@ afterEach(() => { vi.resetAllMocks(); }); -function mountLargeValues(databaseType: DatabaseType, result: Ref) { +function mountLargeValues(databaseType: DatabaseType, result: Ref, withEditor = false) { const toast = vi.fn(); const formatCell = (value: unknown, columnIndex?: number, originalBytes?: number) => binaryCellDisplayText(value, result.value.column_types?.[columnIndex ?? 0], originalBytes, databaseType) ?? String(value ?? ""); const item = (rowId = 1) => ({ id: rowId, sourceIndex: rowId - 1, data: result.value.rows[rowId - 1]!, isNew: false, isDeleted: false, isDirtyCol: [false, false] }); let largeValues: ReturnType | undefined; + let editor: ReturnType | undefined; const app = createApp( defineComponent({ setup() { @@ -81,6 +91,16 @@ function mountLargeValues(databaseType: DatabaseType, result: Ref) cloneRow: vi.fn(), cloneRows: vi.fn(), }); + if (withEditor) editor = useDataGridEditor({ + result: computed(() => result.value), editable: computed(() => true), + databaseType: computed(() => databaseType), connectionId: computed(() => `${databaseType}-1`), database: computed(() => "MAXIMO"), + tableMeta: computed(() => ({ schema: "public", tableName: "APP_DATA", columns: [{ name: "ID", data_type: "NUMBER" }, { name: "APP", data_type: "CLOB" }], primaryKeys: ["ID"] })), + sourceColumns: computed(() => ["ID", "APP"]), onExecuteSql: computed(() => undefined), sql: computed(() => undefined), + searchText: ref(""), dataGridQuickEntryEnabled: computed(() => false), whereFilterInput: ref(""), currentWhereInput: computed(() => undefined), + orderByInput: ref(""), rowStatusFilter: ref("all"), confirmDangerousRowDeletion: computed(() => false), pageSize: ref(100), currentPage: ref(1), + getRowItem: (rowId) => ({ id: rowId, sourceIndex: rowId, data: result.value.rows[rowId]!, isNew: false, isDeleted: false, isDirtyCol: [false, false], status: "normal" }), + prepareSaveBaseline: (changes) => largeValues!.prepareSaveBaseline(changes), emit: vi.fn(), + }); return () => h("div"); }, }), @@ -88,7 +108,7 @@ function mountLargeValues(databaseType: DatabaseType, result: Ref) app.mount(document.createElement("div")); disposers.push(() => app.unmount()); if (!largeValues) throw new Error("useDataGridLargeValues was not mounted"); - return { largeValues, toast, item }; + return { largeValues, toast, item, editor }; } function previewResult(columnType = "bytea", value = "\\x00017f80ff...", originalBytes = 9645) { @@ -103,6 +123,41 @@ function previewResult(columnType = "bytea", value = "\\x00017f80ff...", origina } describe("useDataGridLargeValues", () => { + it.each(["preview", "failed preparation"])("keeps editor edits, deletes and undo after baseline reads during %s", async (action) => { + const result = previewResult("CLOB", "preview", 1); + result.value.rows.push([2, "second preview"]); + result.value.large_value_context = { connectionId: "oracle-1", database: "MAXIMO" }; + result.value.large_value_cells = [{ row_index: 0, column_index: 1, original_bytes: 1, value_ref: "first" }, { row_index: 1, column_index: 1, original_bytes: 1, value_ref: "second" }]; + const { editor } = mountLargeValues("oracle", result, true); + editor!.newRows.value = [[3, "pending insert"]]; + editor!.applyCellValue(0, 1, "pending edit"); + editor!.applyDeleteRow(1); + const rows = result.value.rows; + const first = rows[0]; + const second = rows[1]; + mocks.readLargeValueChunk.mockResolvedValue({ status: "ok", data: "complete", next_offset: 8, eof: true, value_kind: "text" }); + if (action === "preview") { + mocks.prepareDataGridSave.mockResolvedValue({ statements: ["UPDATE APP_DATA", "DELETE APP_DATA"], rollbackStatements: [] }); + await editor!.previewChanges(); + } else { + mocks.prepareDataGridSave.mockRejectedValue(new Error("preparation failed")); + await editor!.saveChanges(); + } + await nextTick(); + expect(editor!.dirtyRows.value.get(0)?.get(1)).toBe("pending edit"); + expect(editor!.deletedRows.value.has(1)).toBe(true); + expect(editor!.newRows.value).toEqual([[3, "pending insert"]]); + expect(editor!.hasPendingChanges.value).toBe(true); + expect(editor!.canUndoPendingChange.value).toBe(true); + expect(result.value.rows).toBe(rows); + expect(result.value.rows[0]).toBe(first); + expect(result.value.rows[1]).toBe(second); + expect(first![1]).toBe("complete"); + expect(mocks.executeBatch).not.toHaveBeenCalled(); + editor!.undoPendingChange(); + expect(editor!.deletedRows.value.has(1)).toBe(false); + expect(editor!.dirtyRows.value.get(0)?.get(1)).toBe("pending edit"); + }); it("loads original LOB baselines before clear/delete SQL generation without replacing pending edits", async () => { const result = previewResult("CLOB", "preview", 1); result.value.large_value_context = { connectionId: "oracle-1", database: "MAXIMO", clientSessionId: "original-session" }; diff --git a/apps/desktop/src/composables/useDataGridLargeValues.ts b/apps/desktop/src/composables/useDataGridLargeValues.ts index 9bca2b9f4a..9d5f25d4d5 100644 --- a/apps/desktop/src/composables/useDataGridLargeValues.ts +++ b/apps/desktop/src/composables/useDataGridLargeValues.ts @@ -662,17 +662,17 @@ export function useDataGridLargeValues(options: UseDataGridLargeValuesOptions) { if (!isCurrent()) throw new Error("LOB result context changed"); // Other consumers can hydrate another cell while these requests await. // Merge only the requested baselines into the latest rows and budget. - const mergedRows = source.rows.map((row) => [...row]); const mergedSizes = new Map(materializedSnapshotBytes.get(source)); for (const cell of cells) { - mergedRows[cell.row_index]![cell.column_index] = rows[cell.row_index]![cell.column_index]; + if (!source.rows[cell.row_index] || cell.column_index >= source.rows[cell.row_index]!.length) throw new Error("LOB result column is unavailable"); const key = largeValueCellKey(cell.row_index, cell.column_index); mergedSizes.set(key, sizes.get(key)!); } if ([...mergedSizes.values()].reduce((sum, size) => sum + size, 0) > 64 * 1024 * 1024) throw new Error("LOB save baseline exceeds the 64 MiB view/edit limit; reload and edit a smaller selection"); source.large_value_refs = [...new Set([...(source.large_value_refs ?? []), ...cells.map((cell) => cell.value_ref!)])]; const resolved = new Set(cells.map((cell) => largeValueCellKey(cell.row_index, cell.column_index))); - source.rows = mergedRows; + // Preserve row identity so the editor keeps pending edits, deletes and undo history. + for (const cell of cells) source.rows[cell.row_index]![cell.column_index] = rows[cell.row_index]![cell.column_index]; source.large_value_cells = source.large_value_cells?.filter((cell) => !resolved.has(largeValueCellKey(cell.row_index, cell.column_index))); materializedSnapshotBytes.set(source, mergedSizes); options.largeValueResolutionVersion.value += 1; diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index b26a8f5165..5df3801ec9 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -2093,10 +2093,10 @@ 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 ?? [])); + const releaseValues = releaseResultLargeValues(tab.result, ...(tab.results ?? [])); releaseTabResultObjectPayloads(tab); clearResultPayload(tab); - await closeSession; + await Promise.all([closeSession, releaseValues]); return true; } diff --git a/crates/dbx-core/assets/agent-protocol-v1.json b/crates/dbx-core/assets/agent-protocol-v1.json index b363666a9e..af91085bbf 100644 --- a/crates/dbx-core/assets/agent-protocol-v1.json +++ b/crates/dbx-core/assets/agent-protocol-v1.json @@ -32,6 +32,8 @@ "execute_query_page", "fetch_query_page", "close_query_session", + "read_large_value_chunk", + "release_large_value", "start_table_read", "fetch_table_read_page", "close_table_read_session", diff --git a/crates/dbx-driver-agent/assets/agent-protocol-v2.json b/crates/dbx-driver-agent/assets/agent-protocol-v2.json index 8cd5f4e230..b0f3ae3046 100644 --- a/crates/dbx-driver-agent/assets/agent-protocol-v2.json +++ b/crates/dbx-driver-agent/assets/agent-protocol-v2.json @@ -36,6 +36,8 @@ "execute_query_page", "fetch_query_page", "close_query_session", + "read_large_value_chunk", + "release_large_value", "start_table_read", "fetch_table_read_page", "close_table_read_session", diff --git a/crates/dbx-driver-agent/src/agent_driver.rs b/crates/dbx-driver-agent/src/agent_driver.rs index e04adb2216..edd8508a3d 100644 --- a/crates/dbx-driver-agent/src/agent_driver.rs +++ b/crates/dbx-driver-agent/src/agent_driver.rs @@ -1229,6 +1229,8 @@ pub enum AgentMethod { ExecuteQueryPage, FetchQueryPage, CloseQuerySession, + ReadLargeValueChunk, + ReleaseLargeValue, StartTableRead, FetchTableReadPage, CloseTableReadSession, @@ -1274,6 +1276,8 @@ impl AgentMethod { Self::ExecuteQueryPage, Self::FetchQueryPage, Self::CloseQuerySession, + Self::ReadLargeValueChunk, + Self::ReleaseLargeValue, Self::StartTableRead, Self::FetchTableReadPage, Self::CloseTableReadSession, @@ -1324,6 +1328,8 @@ impl AgentMethod { Self::StartTableRead => "start_table_read", Self::FetchTableReadPage => "fetch_table_read_page", Self::CloseTableReadSession => "close_table_read_session", + Self::ReadLargeValueChunk => "read_large_value_chunk", + Self::ReleaseLargeValue => "release_large_value", Self::GetExplainInfo => "get_explain_info", Self::ExecuteBatch => "execute_batch", Self::ExecuteTransaction => "execute_transaction", diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index e0081a0fc7..7a6f0eabf2 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -31,6 +31,7 @@ tauri-build = { version = "2.5.6", features = [] } [dependencies] serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +tempfile = "3" serde_yaml_ng = "0.10" base64 = "0.22" log = "0.4" @@ -85,7 +86,6 @@ sysinfo = { version = "0.32", features = ["system"] } [dev-dependencies] dbx-core = { path = "../crates/dbx-core", default-features = false, features = ["test-support"] } tauri = { version = "2.10.3", features = ["test"] } -tempfile = "3" [target.'cfg(target_os = "macos")'.dependencies] objc2 = { version = "0.6.4", default-features = false, features = ["std"] }