mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 12:41:39 +00:00
fix(oceanbase): preserve pooled LOB lifecycle and pending grid changes
This commit is contained in:
efeire
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
+9
-2
@@ -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);
|
||||
|
||||
+104
@@ -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<String> 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<String> 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> T proxy(Class<T> type, java.lang.reflect.InvocationHandler handler) {
|
||||
return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[]{type}, handler));
|
||||
}
|
||||
}
|
||||
@@ -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<QueryResult>) {
|
||||
function mountLargeValues(databaseType: DatabaseType, result: Ref<QueryResult>, 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<typeof useDataGridLargeValues> | undefined;
|
||||
let editor: ReturnType<typeof useDataGridEditor> | undefined;
|
||||
const app = createApp(
|
||||
defineComponent({
|
||||
setup() {
|
||||
@@ -81,6 +91,16 @@ function mountLargeValues(databaseType: DatabaseType, result: Ref<QueryResult>)
|
||||
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<QueryResult>)
|
||||
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" };
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
Reference in New Issue
Block a user