diff --git a/cmd/optional-driver-agent/provider_hive.go b/cmd/optional-driver-agent/provider_hive.go new file mode 100644 index 000000000..2f91fd410 --- /dev/null +++ b/cmd/optional-driver-agent/provider_hive.go @@ -0,0 +1,12 @@ +//go:build gonavi_hive_driver + +package main + +import "GoNavi-Wails/internal/db" + +func init() { + agentDriverType = "hive" + agentDatabaseFactory = func() db.Database { + return &db.HiveDB{} + } +} diff --git a/frontend/public/db-icons/hive.svg b/frontend/public/db-icons/hive.svg new file mode 100644 index 000000000..87677e7f7 --- /dev/null +++ b/frontend/public/db-icons/hive.svg @@ -0,0 +1,7 @@ + + Apache Hive + + + + + diff --git a/frontend/src/components/connectionModal/connectionModalUriParse.ts b/frontend/src/components/connectionModal/connectionModalUriParse.ts index cc453ba25..cf8417dc7 100644 --- a/frontend/src/components/connectionModal/connectionModalUriParse.ts +++ b/frontend/src/components/connectionModal/connectionModalUriParse.ts @@ -92,7 +92,7 @@ export const parseUriToValues = ( parsed.params.get("topology") || "", ).toLowerCase(); const tlsValue = String( - parsed.params.get("tls") || parsed.params.get("useSSL") || "", + parsed.params.get("tls") || parsed.params.get("useSSL") || parsed.params.get("ssl") || "", ) .trim() .toLowerCase(); diff --git a/frontend/src/components/connectionModal/connectionModalUriSchemes.ts b/frontend/src/components/connectionModal/connectionModalUriSchemes.ts index 126d28237..f60f8deca 100644 --- a/frontend/src/components/connectionModal/connectionModalUriSchemes.ts +++ b/frontend/src/components/connectionModal/connectionModalUriSchemes.ts @@ -61,13 +61,31 @@ export const parseMultiHostUri = (uriText: string, expectedScheme: string) => { .map((item) => item.trim()) .filter(Boolean); + const params = new URLSearchParams(queryText); + if (pathText.includes(";")) { + const [dbSegment, ...semiSegments] = pathText.split(";"); + if (semiSegments.some((segment) => segment.includes("="))) { + pathText = dbSegment; + for (const segment of semiSegments) { + const eqIdx = segment.indexOf("="); + if (eqIdx > 0) { + const k = safeDecode(segment.slice(0, eqIdx).trim()); + const v = safeDecode(segment.slice(eqIdx + 1).trim()); + if (k) { + params.set(k, v); + } + } + } + } + } + return { username, password, hosts, database: safeDecode(pathText), hasExplicitPath, - params: new URLSearchParams(queryText), + params, }; }; diff --git a/frontend/src/utils/dataSourceRegistry/hive.test.ts b/frontend/src/utils/dataSourceRegistry/hive.test.ts new file mode 100644 index 000000000..43ca8113f --- /dev/null +++ b/frontend/src/utils/dataSourceRegistry/hive.test.ts @@ -0,0 +1,101 @@ +import { describe, expect, it } from 'vitest'; + +import { buildUriFromValues, getUriPlaceholder, parseUriToValues } from '../../components/connectionModal/connectionModalUri'; +import { buildViewsMetadataQuerySpecs } from '../../components/sidebar/sidebarMetadataQuerySpecs'; +import { supportsTableClearAction } from '../../components/tableDataDangerActions'; +import { CONNECTION_TYPE_GROUPS, getConnectionTypeDefaultPort } from '../connectionTypeCatalog'; +import { findPotentiallyMutatingConnectionStatements, supportsConnectionReadOnlyMode } from '../connectionReadOnly'; +import { getDataSourceCapabilities, getDataSourceCapabilityContract } from '../dataSourceCapabilities'; +import { buildPaginatedSelectSQL, quoteIdentPart } from '../sql'; +import { isMysqlFamilyDialect, resolveSqlDialect } from '../sqlDialectCore'; +import { formatSqlExecutionError } from '../sqlErrorSemantics'; +import { resolveSidebarMetadataDialect } from '../sidebarMetadata'; +import { + canonicalizeDataSourceType, + getDataSourceSpec, + getDataSourceVariants, + getDefaultDataSourceVariant, + supportsAutoDataSourceVariant, +} from './index'; + +const hive = { type: 'hive' } as never; + +describe('Apache Hive registry behavior', () => { + it('joins the big data group on port 10000 with its own dialect and MySQL-family quoting', () => { + const group = CONNECTION_TYPE_GROUPS.find((item) => item.labelKey.endsWith('.bigdata')); + expect(group?.items.map((item) => item.key)).toContain('hive'); + expect(getConnectionTypeDefaultPort('hive')).toBe(10000); + expect(canonicalizeDataSourceType('Apache-Hive')).toBe('hive'); + expect(canonicalizeDataSourceType('hiveserver2')).toBe('hive'); + expect(canonicalizeDataSourceType('hs2')).toBe('hive'); + expect(resolveSqlDialect('hive')).toBe('hive'); + expect(isMysqlFamilyDialect('hive')).toBe(true); + expect(quoteIdentPart('hive', 'ods_orders')).toBe('`ods_orders`'); + expect(buildPaginatedSelectSQL('hive', 'SELECT * FROM `dw`.`ods_orders`', ' ORDER BY `order_id`', 50, 100)).toBe( + 'SELECT * FROM `dw`.`ods_orders` ORDER BY `order_id` LIMIT 50 OFFSET 100', + ); + }); + + it('declares v3 and v4 driver variants with auto detection', () => { + expect(getDataSourceSpec('hive')?.ui?.defaultUser).toBe('hive'); + expect(supportsAutoDataSourceVariant('hive')).toBe(true); + expect(getDefaultDataSourceVariant('hive')).toBe('auto'); + expect(getDataSourceVariants('hive').map((item) => item.id)).toEqual(['v3', 'v4']); + }); + + it('parses hive:// and jdbc:hive2:// connection strings including semicolon parameters', () => { + expect(parseUriToValues('hive://hive:secret@hs2.example.com:10000/dw?auth=PLAIN', 'hive')).toMatchObject({ + host: 'hs2.example.com', + port: 10000, + user: 'hive', + password: 'secret', + database: 'dw', + connectionParams: 'auth=PLAIN', + }); + expect(parseUriToValues('jdbc:hive2://hs2.example.com:10000/dw;auth=noSasl;ssl=true', 'hive')).toMatchObject({ + host: 'hs2.example.com', + port: 10000, + database: 'dw', + useSSL: true, + sslMode: 'required', + }); + expect(buildUriFromValues({ type: 'hive', host: '10.0.0.8', port: 10000, user: 'hive', database: 'dw', timeout: 30 })).toBe( + 'hive://hive@10.0.0.8:10000/dw?timeout=30', + ); + expect(getUriPlaceholder('hive')).toContain('hive://user:pass@127.0.0.1:10000'); + }); + + it('lists views with SHOW VIEWS and exposes Hive capability profile', () => { + const sqls = buildViewsMetadataQuerySpecs(resolveSidebarMetadataDialect('hive'), 'dw').map((spec) => spec.sql); + expect(sqls).toEqual(['SHOW VIEWS IN `dw`', 'SHOW VIEWS']); + const contract = getDataSourceCapabilityContract(hive); + expect(contract.transaction?.supported).toBe(false); + const capabilities = getDataSourceCapabilities(hive); + expect(capabilities.supportsUserManagement).toBe(false); + expect(capabilities.supportsExplainDiagnosis).toBe(true); + expect(capabilities.supportsSqlQueryExport).toBe(true); + expect(supportsTableClearAction('hive')).toBe(true); + }); + + it('explains Hive SemanticException and ParseException messages', () => { + expect(formatSqlExecutionError("Error while compiling statement: FAILED: SemanticException [Error 10001]: Line 1:14 Table not found 'nope'")).toContain( + 'Table or object does not exist', + ); + expect( + formatSqlExecutionError("Error while compiling statement: FAILED: SemanticException [Error 10004]: Line 1:7 Invalid table alias or column reference 'nope'"), + ).toContain('Column does not exist'); + expect(formatSqlExecutionError("Error while compiling statement: FAILED: ParseException line 1:7 cannot recognize input near 'FORM' 't' ''")).toContain( + 'SQL syntax error', + ); + }); + + it('classifies writes for read-only protection', () => { + expect(supportsConnectionReadOnlyMode(hive)).toBe(true); + for (const statement of ['SELECT * FROM dw.ods_orders', 'SHOW TABLES', 'DESCRIBE FORMATTED dw.ods_orders']) { + expect(findPotentiallyMutatingConnectionStatements(hive, statement), statement).toEqual([]); + } + for (const statement of ['INSERT INTO dw.ods_orders VALUES (1)', 'DROP TABLE dw.ods_orders', 'CREATE TABLE dw.t AS SELECT 1 AS id']) { + expect(findPotentiallyMutatingConnectionStatements(hive, statement), statement).toEqual([statement]); + } + }); +}); diff --git a/frontend/src/utils/dataSourceRegistry/uriScheme.ts b/frontend/src/utils/dataSourceRegistry/uriScheme.ts index 78d9c0787..2f0cceb8c 100644 --- a/frontend/src/utils/dataSourceRegistry/uriScheme.ts +++ b/frontend/src/utils/dataSourceRegistry/uriScheme.ts @@ -29,7 +29,11 @@ export const withRegistryUriScheme = (type: unknown, uri: string): string => { /** 连接串使用了描述表声明的哪个 scheme，供借用家族解析的分支（如 MySQL 多主机解析）追加识别。 */ export const findRegistryUriScheme = (type: unknown, uri: unknown): string | undefined => { const text = normalize(uri); - return listRegistryUriSchemes(type).find((scheme) => text.startsWith(`${scheme}://`)); + for (const scheme of listRegistryUriSchemes(type)) { + if (text.startsWith(`${scheme}://`)) return scheme; + if (text.startsWith(`jdbc:${scheme}://`)) return `jdbc:${scheme}`; + } + return undefined; }; /** 使用 Trino 连接表单的类型（Trino 本身与 ui.layout 为 trino 的描述表类型，如 Presto）：默认库是 catalog.schema， diff --git a/frontend/src/utils/sqlErrorSemantics.ts b/frontend/src/utils/sqlErrorSemantics.ts index 9fa0641ad..9593f271b 100644 --- a/frontend/src/utils/sqlErrorSemantics.ts +++ b/frontend/src/utils/sqlErrorSemantics.ts @@ -40,6 +40,8 @@ const SQL_ERROR_RULES: SqlErrorSemanticRule[] = [ /you have an error in your sql syntax/i, // Presto / Trino：line 1:8: mismatched input 'FORM' /mismatched input/i, + // Hive：ParseException line 1:7 cannot recognize input near 'FORM' + /cannot recognize input near/i, ], }, { @@ -56,6 +58,9 @@ const SQL_ERROR_RULES: SqlErrorSemanticRule[] = [ /object\s+.+does not exist/i, // Presto / Trino：Table 'hive.sales.orders' does not exist、Schema / Catalog … does not exist /\b(?:table|view|schema|catalog)\s+.+does not exist/i, + // Hive：Table not found 'nope'、Database does not exist: nope + /\b(?:table|view|database)\s+not\s+found\b/i, + /\bdatabase\s+does\s+not\s+exist\b/i, ], }, { @@ -82,6 +87,8 @@ const SQL_ERROR_RULES: SqlErrorSemanticRule[] = [ /no such column/i, // Presto / Trino：Column 'amount' cannot be resolved /column\s+.+cannot be resolved/i, + // Hive：Invalid table alias or column reference 'amount' + /invalid table alias or column reference/i, ], }, { diff --git a/internal/app/datasource_registry_bridge.go b/internal/app/datasource_registry_bridge.go index ab1ea7fc0..3dac07638 100644 --- a/internal/app/datasource_registry_bridge.go +++ b/internal/app/datasource_registry_bridge.go @@ -162,7 +162,7 @@ func registryUsesBooleanKeywords(dbType string) bool { return false } switch spec.Type { - case "questdb", "greptimedb", "firebird": + case "questdb", "greptimedb", "firebird", "hive": return true default: return false diff --git a/internal/app/methods_file_export_sql_dump.go b/internal/app/methods_file_export_sql_dump.go index d19c4c099..3646ca025 100644 --- a/internal/app/methods_file_export_sql_dump.go +++ b/internal/app/methods_file_export_sql_dump.go @@ -145,7 +145,7 @@ func escapeSQLLiteral(value string) string { func dialectEscapesBackslashInStringLiteral(dbType string) bool { switch strings.ToLower(strings.TrimSpace(dbType)) { case "mysql", "mariadb", "tidb", "oceanbase", "diros", "doris", "starrocks", - "clickhouse", "tdengine", "taos", "greptimedb": + "clickhouse", "tdengine", "taos", "greptimedb", "hive": return true default: return false diff --git a/internal/app/methods_file_export_sql_header.go b/internal/app/methods_file_export_sql_header.go index cf80aac02..0e951c2cd 100644 --- a/internal/app/methods_file_export_sql_header.go +++ b/internal/app/methods_file_export_sql_header.go @@ -22,7 +22,7 @@ func quoteIdentByType(dbType string, ident string) string { } switch dbType { - case "mysql", "mariadb", "oceanbase", "diros", "starrocks", "sphinx", "tdengine", "clickhouse": + case "mysql", "mariadb", "oceanbase", "diros", "starrocks", "sphinx", "tdengine", "clickhouse", "greptimedb", "hive": return "`" + strings.ReplaceAll(ident, "`", "``") + "`" case "kingbase": return db.QuoteKingbaseIdentifier(ident) diff --git a/internal/datasource/specs/hive.json b/internal/datasource/specs/hive.json new file mode 100644 index 000000000..1857738b3 --- /dev/null +++ b/internal/datasource/specs/hive.json @@ -0,0 +1,94 @@ +{ + "type": "hive", + "aliases": [ + "apache-hive", + "hiveserver2", + "hs2" + ], + "displayName": "Apache Hive", + "group": "bigdata", + "order": 10, + "defaultPort": 10000, + "wire": "thrift", + "dialect": "hive", + "family": "mysql", + "objectKind": "table", + "protection": true, + "excelImport": true, + "sqlFileImport": true, + "versionQuery": "SELECT version() AS version", + "sync": { + "source": true, + "target": false, + "model": "relational" + }, + "agent": { + "buildTag": "gonavi_hive_driver", + "version": "1.0.0", + "sourcePrefixes": [ + "hive", + "registry" + ] + }, + "variants": { + "auto": true, + "default": "auto", + "items": [ + { + "id": "v3", + "label": "Hive 3.x", + "descriptionKey": "datasource.variant.hive.v3", + "minServer": "3.0", + "maxServer": "4.0" + }, + { + "id": "v4", + "label": "Hive 4.x +", + "descriptionKey": "datasource.variant.hive.v4", + "minServer": "4.0" + } + ] + }, + "ui": { + "defaultUser": "hive", + "usernameOptional": true, + "formatDdl": false, + "uriSchemes": [ + "hive", + "hive2", + "hiveserver2" + ], + "hiddenObjectGroups": [ + "routines", + "triggers", + "sequences", + "events", + "packages" + ], + "metadataQueries": { + "views": [ + "SHOW VIEWS IN `{{database}}`", + "SHOW VIEWS" + ] + }, + "objectStatements": { + "renameTable": "ALTER TABLE {old} RENAME TO {new}", + "dropTable": "DROP TABLE {table}", + "dropView": "DROP VIEW {view}", + "truncateTable": "TRUNCATE TABLE {table}", + "clearTable": "TRUNCATE TABLE {table}" + }, + "quoting": "backtick", + "pagination": "limit-offset", + "ssl": true, + "sslCAPath": true, + "sslClientCert": true, + "connectionParams": true, + "hint": "Data warehouse / HiveServer2 Thrift", + "icon": { + "color": "#F5A623", + "text": "Hv", + "scale": 0.78 + } + } +} diff --git a/internal/db/data_source_capability_contract.json b/internal/db/data_source_capability_contract.json index 5e1e24344..368f82723 100644 --- a/internal/db/data_source_capability_contract.json +++ b/internal/db/data_source_capability_contract.json @@ -1,6 +1,7 @@ { "version": 2, "profiles": { + "sql-hive": {"query": {"supported": true}, "metadata": {"supported": true}, "transaction": {"supported": false, "reason": "transaction_not_supported", "alternative": "execute-statement-directly", "messageKey": "query_editor.message.unsupported_source"}, "pagination": {"supported": true}, "cancel": {"supported": true, "runtimeProbe": true}, "schema": {"supported": false, "reason": "schema_designer_not_supported", "alternative": "use-database-metadata", "messageKey": "query_editor.message.unsupported_source"}, "sampling": {"supported": true}, "streaming": {"supported": false, "reason": "not_streaming_data_source", "alternative": "use-paginated-query", "messageKey": "query_editor.message.unsupported_source"}, "dangerousOperations": {"supported": true}, "ui": {"forceReadOnlyQueryResult": true, "forceReadOnlyStructureDesigner": true, "explainDiagnosis": true, "sqlQueryExport": true, "copyInsert": true, "mockData": true, "createDatabase": true, "dropDatabase": true, "parameterBinding": true}, "navigation": {"primaryVisibilitySupported": true, "primaryKind": "database", "secondarySchemaVisibilitySupported": false, "schemaIdentifierCaseSensitive": false}}, "sql-gbase8s": {"query": {"supported": true}, "metadata": {"supported": true}, "transaction": {"supported": true}, "pagination": {"supported": true}, "cancel": {"supported": true, "runtimeProbe": true}, "schema": {"supported": true}, "sampling": {"supported": true}, "streaming": {"supported": false, "reason": "not_streaming_data_source", "alternative": "use-paginated-query", "messageKey": "query_editor.message.unsupported_source"}, "dangerousOperations": {"supported": true}, "ui": {"userManagement": false, "parameterBinding": false, "explainDiagnosis": false, "sqlQueryExport": true, "copyInsert": true, "mockData": true, "copyTable": false, "createDatabase": false, "createDatabaseCharset": false, "dropDatabase": false}, "navigation": {"primaryVisibilitySupported": true, "primaryKind": "database", "secondarySchemaVisibilitySupported": false, "schemaIdentifierCaseSensitive": false}}, "sql-gbase8a": {"query": {"supported": true}, "metadata": {"supported": true}, "transaction": {"supported": true}, "pagination": {"supported": true}, "cancel": {"supported": true, "runtimeProbe": true}, "schema": {"supported": true}, "sampling": {"supported": true}, "streaming": {"supported": false, "reason": "not_streaming_data_source", "alternative": "use-paginated-query", "messageKey": "query_editor.message.unsupported_source"}, "dangerousOperations": {"supported": true}, "ui": {"userManagement": false, "parameterBinding": true, "explainDiagnosis": true, "sqlQueryExport": true, "copyInsert": true, "mockData": true, "copyTable": true, "createDatabase": true, "createDatabaseCharset": true, "dropDatabase": true}, "navigation": {"primaryVisibilitySupported": true, "primaryKind": "database", "secondarySchemaVisibilitySupported": false, "schemaIdentifierCaseSensitive": false}}, "sql-yashandb": {"query": {"supported": true}, "metadata": {"supported": true}, "transaction": {"supported": true}, "pagination": {"supported": true}, "cancel": {"supported": true, "runtimeProbe": true}, "schema": {"supported": true}, "sampling": {"supported": true}, "streaming": {"supported": false, "reason": "not_streaming_data_source", "alternative": "use-paginated-query", "messageKey": "query_editor.message.unsupported_source"}, "dangerousOperations": {"supported": true}, "ui": {"userManagement": false, "parameterBinding": true, "explainDiagnosis": true, "sqlQueryExport": true, "copyInsert": true, "mockData": true, "preferManualTotalCount": true, "supportsApproximateTableCount": true}, "navigation": {"primaryVisibilitySupported": true, "primaryKind": "owner", "secondarySchemaVisibilitySupported": false, "schemaIdentifierCaseSensitive": false}}, @@ -1497,6 +1498,7 @@ } }, "drivers": { + "hive": "sql-hive", "firebird": "sql-firebird", "yashandb": "sql-yashandb", "gbase8s": "sql-gbase8s", diff --git a/internal/db/driver_agent_revisions_gen.go b/internal/db/driver_agent_revisions_gen.go index a43bb6452..7b31f4ddd 100644 --- a/internal/db/driver_agent_revisions_gen.go +++ b/internal/db/driver_agent_revisions_gen.go @@ -4,49 +4,50 @@ package db func init() { optionalDriverAgentRevisions = map[string]string{ - "mariadb": "src-47e754c5477d8c5c", - "oceanbase": "src-2dfbdb0ec1118342", - "diros": "src-84271b9877a878a2", - "starrocks": "src-051943dd82739ccd", - "sphinx": "src-5f6c088bb33c7afc", - "sqlserver": "src-05b2713c868ec747", - "sqlite": "src-7c0ef9af425e6e5f", - "duckdb": "src-4609724c8fdaa0ff", - "dameng": "src-ec287b1dbce72cd2", - "kingbase": "src-394196650cc74985", - "highgo": "src-ffe45b7292745afa", - "vastbase": "src-303fed14c8a73790", - "opengauss": "src-a9717c58f6e9dacf", - "gaussdb": "src-f5de71344da4234f", - "iris": "src-9fca5bec64569c0d", - "cache": "src-f45f09af7574a31a", - "mongodb": "src-d2c7bf5ad5eaef25", - "tdengine": "src-35ce749d51e600dd", - "iotdb": "src-b679b15ec9b2a00d", - "clickhouse": "src-f0f299fd19299d30", - "elasticsearch": "src-24a65eee75612650", - "trino": "src-3f6e68269c9f3071", - "kafka": "src-018af0dfaf84cfd5", - "rocketmq": "src-de65a93400d7304f", - "pulsar": "src-8a48137f26982d59", - "cockroachdb": "src-09e7c44631d9d449", - "etcd": "src-3986b1d4209c5a79", - "firebird": "src-e373e5ce82dad42f", - "gbase8a": "src-fe2d5e1926e8b75d", - "gbase8c": "src-bf76118076fd55ac", - "gbase8s": "src-388bf81a7fc10e87", - "greptimedb": "src-4e86d44b261672ba", - "influxdb": "src-64b6cd7c6af487d3", - "kwdb": "src-d4dd9d7c8491b7e0", - "meilisearch": "src-11d58641ef95bc17", - "opensearch": "src-cfa1155bd2340343", - "presto": "src-f092f767c09df7b2", - "questdb": "src-ac52fe0d9517ff2d", - "tidb": "src-6e2751312e88fbca", - "timescaledb": "src-e63a3e816b6fa038", - "typesense": "src-3766928647fbf2e3", - "weaviate": "src-802aa3f86493c80b", - "yashandb": "src-52d286249eed49de", - "zookeeper": "src-bd49910a4b234bf1", + "mariadb": "src-24cff0b0cd004883", + "oceanbase": "src-dc9e3f44d8caafab", + "diros": "src-28223cf02d14100b", + "starrocks": "src-38a128dc1b50ca54", + "sphinx": "src-c538ef14dd64a40c", + "sqlserver": "src-e9c6c0e9a9c52bc4", + "sqlite": "src-37bcd2b4d75acfb4", + "duckdb": "src-2ffa40f69d925e57", + "dameng": "src-0b967b8da063ee87", + "kingbase": "src-8480fbd0e09ba859", + "highgo": "src-dec92de8bbcc4d66", + "vastbase": "src-602bc2581f4cea81", + "opengauss": "src-6814ac3e6bea9624", + "gaussdb": "src-cc7d0d36310300ed", + "iris": "src-9e3cff0db0f702d4", + "cache": "src-1786fdb7306ec582", + "mongodb": "src-c22c9fc1b0cfbfc2", + "tdengine": "src-d6acb6ef39fee3d5", + "iotdb": "src-e04624a4eb5a57ff", + "clickhouse": "src-14c6ab3d1f166779", + "elasticsearch": "src-288b8d6834473ab6", + "trino": "src-53750718cddb0291", + "kafka": "src-0e8523f49163f232", + "rocketmq": "src-163ff11e0b98bbd4", + "pulsar": "src-ab58d3bd006e53ca", + "cockroachdb": "src-47b9936b051b5188", + "etcd": "src-8c447a1e7de740a4", + "firebird": "src-3ac03713bf32027d", + "gbase8a": "src-a3daa1ad621497b1", + "gbase8c": "src-7e9d2a0d761eae12", + "gbase8s": "src-fe69da7b44959db4", + "greptimedb": "src-671145b8d3776449", + "hive": "src-c4c6eb11b5d5619f", + "influxdb": "src-fbabe6656f67e3d6", + "kwdb": "src-a8d668b42cde50aa", + "meilisearch": "src-63ac1be1e33ef2dd", + "opensearch": "src-ad3a98f99af600f2", + "presto": "src-2ee5b16533ecc679", + "questdb": "src-09be0c048d81e013", + "tidb": "src-42930ff8006797d1", + "timescaledb": "src-53a3b6f332aa11fc", + "typesense": "src-c19583286cfeae57", + "weaviate": "src-120be62de08d06cd", + "yashandb": "src-cb925fc83f05f259", + "zookeeper": "src-a461a35109794886", } } diff --git a/internal/db/hive_impl.go b/internal/db/hive_impl.go new file mode 100644 index 000000000..98e1e243c --- /dev/null +++ b/internal/db/hive_impl.go @@ -0,0 +1,368 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "context" + "crypto/tls" + "fmt" + "net" + "net/url" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "GoNavi-Wails/internal/connection" + proxytunnel "GoNavi-Wails/internal/proxy" + "GoNavi-Wails/internal/ssh" +) + +const ( + defaultHivePort = 10000 + defaultHiveDatabase = "default" + defaultHiveUser = "hive" +) + +var hiveVersionPattern = regexp.MustCompile(`(\d+\.\d+(?:\.\d+)?)`) + +// HiveDB 通过 HiveServer2 TCLIService Thrift Binary 协议访问 Apache Hive（3.x / 4.x）。 +type HiveDB struct { + driverVariantState + transport *hiveTransport + sessionID hiveHandleID + serverProto int32 + database string + currentDB string + forwarder *ssh.LocalForwarder + timeout time.Duration + execMu sync.Mutex +} + +var ( + _ Database = (*HiveDB)(nil) + _ QueryContexter = (*HiveDB)(nil) + _ ExecContexter = (*HiveDB)(nil) + _ StreamQueryExecer = (*HiveDB)(nil) + _ QueryArgsContexter = (*HiveDB)(nil) + _ ExecArgsContexter = (*HiveDB)(nil) + _ BatchApplierContext = (*HiveDB)(nil) + _ DriverVariantReporter = (*HiveDB)(nil) +) + +type hiveResolvedTarget struct { + host string + port int + user string + password string + database string + useSSL bool + params url.Values +} + +func resolveHiveTarget(config connection.ConnectionConfig) hiveResolvedTarget { + target := hiveResolvedTarget{ + host: strings.TrimSpace(config.Host), + port: config.Port, + user: strings.TrimSpace(config.User), + password: config.Password, + database: strings.TrimSpace(config.Database), + useSSL: config.UseSSL, + params: parseHiveConnectionParams(config.ConnectionParams), + } + if uri := strings.TrimSpace(config.URI); uri != "" { + applyHiveURI(&target, uri) + } + if target.host == "" { + target.host = "127.0.0.1" + } + if target.port <= 0 { + target.port = defaultHivePort + } + if target.user == "" { + target.user = firstNonEmptyParam(target.params, "user", "username") + } + if target.user == "" { + target.user = defaultHiveUser + } + if target.password == "" { + target.password = firstNonEmptyParam(target.params, "password") + } + if target.database == "" { + target.database = firstNonEmptyParam(target.params, "database", "db", "schema") + } + if target.database == "" { + target.database = defaultHiveDatabase + } + if sslParam := strings.ToLower(strings.TrimSpace(firstNonEmptyParam(target.params, "ssl", "useSSL", "tls"))); sslParam != "" { + switch sslParam { + case "true", "1", "required", "skip-verify", "preferred": + target.useSSL = true + } + } + return target +} + +func parseHiveConnectionParams(raw string) url.Values { + values := url.Values{} + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return values + } + normalized := strings.ReplaceAll(trimmed, ";", "&") + parsed, err := url.ParseQuery(normalized) + if err != nil { + return values + } + for k, v := range parsed { + if key := strings.TrimSpace(k); key != "" { + values[key] = v + } + } + return values +} + +func applyHiveURI(target *hiveResolvedTarget, rawURI string) { + text := strings.TrimSpace(rawURI) + if len(text) > len("jdbc:") && strings.EqualFold(text[:len("jdbc:")], "jdbc:") { + text = text[len("jdbc:"):] + } + var semiPart string + if semiIdx := strings.Index(text, ";"); semiIdx >= 0 { + semiPart = text[semiIdx+1:] + text = text[:semiIdx] + } + parsed, err := url.Parse(text) + if err != nil { + return + } + if host := strings.TrimSpace(parsed.Hostname()); host != "" { + target.host = host + } + if portText := strings.TrimSpace(parsed.Port()); portText != "" { + if port, err := strconv.Atoi(portText); err == nil && port > 0 { + target.port = port + } + } + if parsed.User != nil { + if u := strings.TrimSpace(parsed.User.Username()); u != "" && target.user == "" { + target.user = u + } + if p, ok := parsed.User.Password(); ok && target.password == "" { + target.password = p + } + } + if dbPath := strings.Trim(strings.TrimSpace(parsed.Path), "/"); dbPath != "" && target.database == "" { + target.database = dbPath + } + for k, v := range parsed.Query() { + if _, exists := target.params[k]; !exists { + target.params[k] = v + } + } + for k, v := range parseHiveConnectionParams(semiPart) { + if _, exists := target.params[k]; !exists { + target.params[k] = v + } + } +} + +func isHiveNoSASL(params url.Values) bool { + auth := strings.ToLower(strings.TrimSpace(firstNonEmptyParam(params, "auth", "authentication"))) + if auth == "nosasl" || auth == "no_sasl" || auth == "none_nosasl" || auth == "raw" { + return true + } + sasl := strings.ToLower(strings.TrimSpace(firstNonEmptyParam(params, "sasl", "useSasl", "use_sasl"))) + return sasl == "false" || sasl == "0" || sasl == "no" +} + +func buildHiveSessionConfiguration(database string, params url.Values) map[string]string { + conf := map[string]string{ + "set:hiveconf:hive.resultset.use.unique.column.names": "false", + } + if db := strings.TrimSpace(database); db != "" { + conf["use:database"] = db + } + for key, values := range params { + if len(values) == 0 { + continue + } + val := strings.TrimSpace(values[len(values)-1]) + lower := strings.ToLower(strings.TrimSpace(key)) + switch { + case strings.HasPrefix(lower, "hiveconf:"): + conf["set:"+strings.TrimSpace(key)] = val + case strings.HasPrefix(lower, "hivevar:"): + conf["set:"+strings.TrimSpace(key)] = val + case strings.HasPrefix(lower, "set:"): + conf[strings.TrimSpace(key)] = val + case strings.HasPrefix(lower, "hive.") || strings.HasPrefix(lower, "tez.") || strings.HasPrefix(lower, "mapreduce."): + conf["set:hiveconf:"+strings.TrimSpace(key)] = val + } + } + return conf +} + +func (h *HiveDB) Connect(config connection.ConnectionConfig) (err error) { + _ = h.Close() + defer func() { + if err != nil { + _ = h.Close() + } + }() + + target := resolveHiveTarget(config) + h.timeout = getConnectTimeout(config) + h.database = target.database + serverHost := target.host + dialHost := target.host + dialPort := target.port + + if config.UseSSH { + forwarder, sshErr := ssh.AcquireLocalForwarder(config.SSH, target.host, target.port) + if sshErr != nil { + return fmt.Errorf("创建 Hive SSH 隧道失败：%w", sshErr) + } + h.forwarder = forwarder + host, portText, splitErr := net.SplitHostPort(forwarder.LocalAddr) + if splitErr != nil { + return fmt.Errorf("解析本地转发地址失败：%w", splitErr) + } + port, portErr := strconv.Atoi(portText) + if portErr != nil { + return fmt.Errorf("解析本地转发端口失败：%w", portErr) + } + dialHost, dialPort = host, port + } + + ctx, cancel := context.WithTimeout(context.Background(), h.timeout) + defer cancel() + + address := net.JoinHostPort(dialHost, strconv.Itoa(dialPort)) + var rawConn net.Conn + if config.UseProxy { + rawConn, err = proxytunnel.DialContext(ctx, config.Proxy, "tcp", address) + } else { + rawConn, err = (&net.Dialer{KeepAlive: 30 * time.Second}).DialContext(ctx, "tcp", address) + } + if err != nil { + return err + } + + if target.useSSL { + sslConfig := config + sslConfig.UseSSL = true + if strings.TrimSpace(sslConfig.SSLMode) == "" { + sslConfig.SSLMode = "required" + } + tlsCfg, tlsErr := resolveGenericTLSConfig(sslConfig) + if tlsErr != nil { + _ = rawConn.Close() + return tlsErr + } + if tlsCfg == nil { + tlsCfg = &tls.Config{MinVersion: tls.VersionTLS12} + } else { + tlsCfg = tlsCfg.Clone() + } + if tlsCfg.ServerName == "" { + tlsCfg.ServerName = serverHost + } + tlsConn := tls.Client(rawConn, tlsCfg) + if err := tlsConn.HandshakeContext(ctx); err != nil { + _ = rawConn.Close() + return err + } + rawConn = tlsConn + } + + noSASL := isHiveNoSASL(target.params) + framed := !noSASL || strings.EqualFold(firstNonEmptyParam(target.params, "transport", "transportMode"), "framed") + h.transport = newHiveTransport(rawConn, framed) + if !noSASL { + if err := h.transport.handshakeSASLPlain(ctx, target.user, target.password); err != nil { + return err + } + } + + sessionConf := buildHiveSessionConfiguration(target.database, target.params) + sessionID, serverProto, err := h.transport.openSession(ctx, hiveProtocolV11, target.user, target.password, sessionConf) + if err != nil { + return err + } + h.sessionID = sessionID + h.serverProto = serverProto + h.currentDB = target.database + + version, verErr := h.detectVersion(ctx) + if verErr != nil { + return verErr + } + if err := h.resolve("hive", config, version); err != nil { + return err + } + if target.database != "" && !strings.EqualFold(target.database, defaultHiveDatabase) { + if _, err := h.execStatementOnSession(ctx, "USE "+quoteHiveIdent(target.database)); err != nil { + return err + } + h.currentDB = target.database + } + return nil +} + +func (h *HiveDB) detectVersion(ctx context.Context) (string, error) { + rows, _, err := h.queryStatementOnSession(ctx, "SELECT version() AS version", nil) + if err != nil { + return "", err + } + raw := FirstQueryRowValue(rows) + return parseHiveVersion(raw), nil +} + +func parseHiveVersion(banner string) string { + match := hiveVersionPattern.FindStringSubmatch(strings.TrimSpace(banner)) + if len(match) < 2 { + return "" + } + return match[1] +} + +func (h *HiveDB) Close() error { + h.execMu.Lock() + defer h.execMu.Unlock() + var firstErr error + if h.transport != nil { + if len(h.sessionID.guid) > 0 { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + _ = h.transport.closeSession(ctx, h.sessionID) + cancel() + h.sessionID = hiveHandleID{} + } + if err := h.transport.Close(); err != nil && firstErr == nil { + firstErr = err + } + h.transport = nil + } + if h.forwarder != nil { + if err := h.forwarder.Close(); err != nil && firstErr == nil { + firstErr = err + } + h.forwarder = nil + } + return firstErr +} + +func (h *HiveDB) Ping() error { + ctx, cancel := context.WithTimeout(context.Background(), h.timeoutOrDefault()) + defer cancel() + _, _, err := h.QueryContext(ctx, "SELECT 1") + return err +} + +func (h *HiveDB) timeoutOrDefault() time.Duration { + if h.timeout > 0 { + return h.timeout + } + return 15 * time.Second +} diff --git a/internal/db/hive_impl_query.go b/internal/db/hive_impl_query.go new file mode 100644 index 000000000..1e5fa49ef --- /dev/null +++ b/internal/db/hive_impl_query.go @@ -0,0 +1,382 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" +) + +const hiveCancelTimeout = 5 * time.Second + +// normalizeHiveStatement 去掉语句末尾的分号：HiveServer2 Thrift ExecuteStatement 只接受单条不含尾部分号的 HQL。 +func normalizeHiveStatement(query string) string { + return strings.TrimRight(strings.TrimSpace(query), "; \t\r\n") +} + +func (h *HiveDB) QueryContext(ctx context.Context, query string) ([]map[string]interface{}, []string, error) { + h.execMu.Lock() + defer h.execMu.Unlock() + return h.queryStatementOnSession(ctx, query, RowBudgetFromContext(ctx)) +} + +func (h *HiveDB) Query(query string) ([]map[string]interface{}, []string, error) { + return h.QueryContext(metadataContextFor(h), query) +} + +func (h *HiveDB) StreamQueryContext(ctx context.Context, query string, consumer QueryStreamConsumer) error { + h.execMu.Lock() + defer h.execMu.Unlock() + if h.transport == nil { + return localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + sink := &hiveStreamSink{consumer: consumer} + _, err := h.runOperation(ctx, query, sink) + return err +} + +func (h *HiveDB) StreamQuery(query string, consumer QueryStreamConsumer) error { + return h.StreamQueryContext(metadataContextFor(h), query, consumer) +} + +func (h *HiveDB) ExecContext(ctx context.Context, query string) (int64, error) { + h.execMu.Lock() + defer h.execMu.Unlock() + return h.execStatementOnSession(ctx, query) +} + +func (h *HiveDB) Exec(query string) (int64, error) { + return h.ExecContext(metadataContextFor(h), query) +} + +func (h *HiveDB) QueryContextWithArgs(ctx context.Context, query string, args []any) ([]map[string]interface{}, []string, error) { + bound, err := hiveBindArgs(query, args) + if err != nil { + return nil, nil, err + } + return h.QueryContext(ctx, bound) +} + +func (h *HiveDB) ExecContextWithArgs(ctx context.Context, query string, args []any) (int64, error) { + bound, err := hiveBindArgs(query, args) + if err != nil { + return 0, err + } + return h.ExecContext(ctx, bound) +} + +func (h *HiveDB) queryStatementOnSession(ctx context.Context, query string, budget *RowBudget) ([]map[string]interface{}, []string, error) { + if h.transport == nil { + return nil, nil, localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + collector := &hiveRowCollector{budget: budget} + if _, err := h.runOperation(ctx, query, collector); err != nil { + return nil, nil, err + } + if collector.data == nil { + collector.data = make([]map[string]interface{}, 0) + } + return collector.data, collector.names, nil +} + +func (h *HiveDB) execStatementOnSession(ctx context.Context, query string) (int64, error) { + if h.transport == nil { + return 0, localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + return h.runOperation(ctx, query, nil) +} + +func (h *HiveDB) runOperation(ctx context.Context, rawQuery string, sink hiveRowSink) (int64, error) { + statement := normalizeHiveStatement(rawQuery) + if statement == "" { + return 0, nil + } + op, err := h.transport.executeStatement(ctx, h.sessionID, statement, nil, true, 0) + if err != nil { + return 0, err + } + closed := false + closeOp := func() { + if closed || len(op.id.guid) == 0 { + return + } + closed = true + cctx, cancel := context.WithTimeout(context.Background(), hiveCancelTimeout) + defer cancel() + _ = h.transport.closeOperation(cctx, op) + } + defer closeOp() + + cancelOp := func() { + if len(op.id.guid) == 0 { + return + } + cctx, cancel := context.WithTimeout(context.Background(), hiveCancelTimeout) + defer cancel() + _ = h.transport.cancelOperation(cctx, op) + } + + hasResultSet, err := h.waitOperationFinished(ctx, op, cancelOp) + if err != nil { + return 0, err + } + if op.hasResultSet { + hasResultSet = true + } + + var affected int64 + if op.hasModifiedCount && op.modifiedRowCount > 0 { + affected = int64(op.modifiedRowCount) + } + + if !hasResultSet || sink == nil { + return affected, nil + } + + cols, err := h.transport.getResultSetMetadata(ctx, op) + if err != nil { + if ctx.Err() != nil { + cancelOp() + return 0, ctx.Err() + } + return 0, err + } + names := uniqueHiveColumnNames(cols) + if err := sink.onColumns(names); err != nil { + return affected, err + } + + for { + batch, fetchErr := h.transport.fetchResults(ctx, op, hiveFetchNext, defaultHiveFetchBatchSize) + if fetchErr != nil { + if ctx.Err() != nil { + cancelOp() + return 0, ctx.Err() + } + return 0, fetchErr + } + if len(batch) == 0 { + break + } + for _, row := range batch { + if err := sink.onRow(row); err != nil { + if errors.Is(err, errHiveStopFetch) { + cancelOp() + return affected, nil + } + cancelOp() + return 0, err + } + } + } + return affected, nil +} + +func (h *HiveDB) waitOperationFinished(ctx context.Context, op hiveOperationHandle, cancelOp func()) (bool, error) { + pollWait := 20 * time.Millisecond + for { + st, err := h.transport.getOperationStatus(ctx, op) + if err != nil { + if ctx.Err() != nil { + cancelOp() + return false, ctx.Err() + } + return false, err + } + switch st.operationState { + case hiveOpFinished: + return st.hasResultSet, nil + case hiveOpInitialized, hiveOpPending, hiveOpRunning: + timer := time.NewTimer(pollWait) + select { + case <-ctx.Done(): + timer.Stop() + cancelOp() + return false, ctx.Err() + case <-timer.C: + if pollWait < 200*time.Millisecond { + pollWait *= 2 + } + } + case hiveOpCanceled: + return false, context.Canceled + case hiveOpTimedout: + return false, context.DeadlineExceeded + case hiveOpError, hiveOpClosed, hiveOpUnknown: + msg := strings.TrimSpace(st.errorMessage) + if msg == "" { + msg = fmt.Sprintf("Hive operation failed in state %d", st.operationState) + } + return false, &hiveQueryError{ + sqlState: st.sqlState, + errorCode: st.errorCode, + message: msg, + } + default: + return st.hasResultSet, nil + } + } +} + +var errHiveStopFetch = errors.New("stop hive fetch") + +type hiveRowCollector struct { + names []string + data []map[string]interface{} + budget *RowBudget +} + +func (c *hiveRowCollector) onColumns(names []string) error { + c.names = names + return nil +} + +func (c *hiveRowCollector) onRow(values []any) error { + if !c.budget.CanMaterializeRow(len(c.data)) { + return errHiveStopFetch + } + row := make(map[string]interface{}, len(c.names)) + for i, name := range c.names { + if i < len(values) { + row[name] = normalizeQueryValue(values[i]) + } else { + row[name] = nil + } + } + row, fieldTruncated := boundQueryRowFields(row, c.budget.MaxFieldBytes()) + if fieldTruncated { + c.budget.MarkFieldTruncated() + } + if !c.budget.ConsumeRow(estimateQueryRowBytes(row)) { + return errHiveStopFetch + } + c.data = append(c.data, row) + return nil +} + +type hiveStreamSink struct { + consumer QueryStreamConsumer + names []string +} + +func (s *hiveStreamSink) onColumns(names []string) error { + s.names = names + return s.consumer.SetColumns(names) +} + +func (s *hiveStreamSink) onRow(values []any) error { + row := make(map[string]interface{}, len(s.names)) + for i, name := range s.names { + if i < len(values) { + row[name] = values[i] + } else { + row[name] = nil + } + } + return s.consumer.ConsumeRow(row) +} + +// hiveBindArgs 将 SQL 中不在引号或注释内的 ? 占位符替换为 HiveQL 字面量。 +func hiveBindArgs(query string, args []any) (string, error) { + if len(args) == 0 { + return query, nil + } + var out strings.Builder + out.Grow(len(query) + len(args)*16) + argIdx := 0 + var quote rune + for i := 0; i < len(query); i++ { + ch := rune(query[i]) + if quote != 0 { + out.WriteByte(query[i]) + if ch == '\\' && i+1 < len(query) { + i++ + out.WriteByte(query[i]) + continue + } + if ch == quote { + if i+1 < len(query) && rune(query[i+1]) == quote { + i++ + out.WriteByte(query[i]) + continue + } + quote = 0 + } + continue + } + if ch == '\'' || ch == '"' || ch == '`' { + quote = ch + out.WriteByte(query[i]) + continue + } + if ch == '-' && i+1 < len(query) && query[i+1] == '-' { + end := strings.IndexByte(query[i:], '\n') + if end < 0 { + out.WriteString(query[i:]) + break + } + out.WriteString(query[i : i+end+1]) + i += end + continue + } + if ch == '/' && i+1 < len(query) && query[i+1] == '*' { + end := strings.Index(query[i+2:], "*/") + if end < 0 { + out.WriteString(query[i:]) + break + } + out.WriteString(query[i : i+2+end+2]) + i += 2 + end + 1 + continue + } + if ch == '?' { + if argIdx >= len(args) { + return "", fmt.Errorf("insufficient arguments for query placeholders: got %d", len(args)) + } + out.WriteString(hiveLiteral(args[argIdx])) + argIdx++ + continue + } + out.WriteByte(query[i]) + } + if argIdx != len(args) { + return "", fmt.Errorf("placeholder count %d does not match argument count %d", argIdx, len(args)) + } + return out.String(), nil +} + +func hiveLiteral(value any) string { + switch v := value.(type) { + case nil: + return "NULL" + case bool: + if v { + return "TRUE" + } + return "FALSE" + case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64: + return fmt.Sprintf("%d", v) + case float32: + return strconv.FormatFloat(float64(v), 'f', -1, 32) + case float64: + return strconv.FormatFloat(v, 'f', -1, 64) + case time.Time: + return "'" + v.Format("2006-01-02 15:04:05") + "'" + case []byte: + return hiveQuoteString(string(v)) + case string: + return hiveQuoteString(v) + default: + return hiveQuoteString(fmt.Sprint(v)) + } +} + +func hiveQuoteString(s string) string { + escaped := strings.ReplaceAll(s, `\`, `\\`) + escaped = strings.ReplaceAll(escaped, `'`, `\'`) + return "'" + escaped + "'" +} diff --git a/internal/db/hive_impl_schema.go b/internal/db/hive_impl_schema.go new file mode 100644 index 000000000..b69272b5d --- /dev/null +++ b/internal/db/hive_impl_schema.go @@ -0,0 +1,387 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "context" + "fmt" + "strings" + + "GoNavi-Wails/internal/connection" +) + +func quoteHiveIdent(name string) string { + cleaned := strings.Trim(strings.TrimSpace(name), "`") + return "`" + strings.ReplaceAll(cleaned, "`", "``") + "`" +} + +func quoteHiveTable(dbName, tableName string) string { + db, table := resolveHiveTableTarget(dbName, tableName, "") + if db == "" { + return quoteHiveIdent(table) + } + return quoteHiveIdent(db) + "." + quoteHiveIdent(table) +} + +func resolveHiveTableTarget(dbName, tableName, defaultDB string) (string, string) { + rawDB := strings.Trim(strings.TrimSpace(dbName), "`") + rawTable := strings.TrimSpace(tableName) + if schema, table := SplitSQLQualifiedNameForDialect(rawTable, "mysql"); table != "" { + if schema != "" { + return strings.Trim(schema, "`"), strings.Trim(table, "`") + } + rawTable = strings.Trim(table, "`") + } else { + rawTable = strings.Trim(rawTable, "`") + } + if rawDB == "" { + rawDB = strings.Trim(strings.TrimSpace(defaultDB), "`") + } + return rawDB, rawTable +} + +func (h *HiveDB) effectiveDatabase(dbName string) string { + if db := strings.Trim(strings.TrimSpace(dbName), "`"); db != "" { + return db + } + if db := strings.Trim(strings.TrimSpace(h.database), "`"); db != "" { + return db + } + return defaultHiveDatabase +} + +func (h *HiveDB) GetDatabases() ([]string, error) { + rows, _, err := h.QueryContext(metadataContextFor(h), "SHOW DATABASES") + if err != nil { + return nil, err + } + databases := make([]string, 0, len(rows)) + for _, row := range rows { + name := rowText(row, "database_name", "namespace", "database") + if name == "" { + name = firstRowValueText(row) + } + if name != "" { + databases = append(databases, name) + } + } + return databases, nil +} + +func (h *HiveDB) GetTables(dbName string) ([]string, error) { + ctx := metadataContextFor(h) + targetDB := h.effectiveDatabase(dbName) + rows, _, err := h.QueryContext(ctx, "SHOW TABLES IN "+quoteHiveIdent(targetDB)) + if err != nil { + return nil, err + } + viewSet := make(map[string]struct{}) + if viewRows, _, viewErr := h.QueryContext(ctx, "SHOW VIEWS IN "+quoteHiveIdent(targetDB)); viewErr == nil { + for _, row := range viewRows { + vName := rowText(row, "tab_name", "view_name", "table_name") + if vName == "" { + vName = firstRowValueText(row) + } + if vName != "" { + viewSet[strings.ToLower(vName)] = struct{}{} + } + } + } + tables := make([]string, 0, len(rows)) + for _, row := range rows { + name := rowText(row, "tab_name", "table_name", "tableName", "name") + if name == "" { + name = firstRowValueText(row) + } + if name == "" { + continue + } + if _, isView := viewSet[strings.ToLower(name)]; isView { + continue + } + tables = append(tables, name) + } + return tables, nil +} + +func (h *HiveDB) GetCreateStatement(dbName, tableName string) (string, error) { + targetDB, targetTable := resolveHiveTableTarget(dbName, tableName, h.database) + if targetTable == "" { + return "", localizedDatabaseRuntimeError("db.backend.error.table_name_required", nil) + } + rows, _, err := h.QueryContext(metadataContextFor(h), "SHOW CREATE TABLE "+quoteHiveTable(targetDB, targetTable)) + if err != nil { + return "", err + } + var lines []string + for _, row := range rows { + rawVal := firstMapValueOf(row, "createtab_stmt", "Create Table", "create_table") + var line string + if rawVal != nil { + line = strings.TrimRight(fmt.Sprint(rawVal), "\r\n") + } + if strings.TrimSpace(line) == "" { + line = firstRowValueText(row) + } + if strings.TrimSpace(line) != "" { + lines = append(lines, line) + } + } + ddl := strings.TrimSpace(strings.Join(lines, "\n")) + if ddl == "" { + return "", localizedDatabaseRuntimeError("db.backend.error.hive_ddl_empty", map[string]any{"name": targetTable}) + } + return ddl, nil +} + +func (h *HiveDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefinition, error) { + targetDB, targetTable := resolveHiveTableTarget(dbName, tableName, h.database) + if targetTable == "" { + return nil, localizedDatabaseRuntimeError("db.backend.error.table_name_required", nil) + } + rows, _, err := h.QueryContext(metadataContextFor(h), "DESCRIBE "+quoteHiveTable(targetDB, targetTable)) + if err != nil { + return nil, err + } + return parseHiveDescribeRows(rows), nil +} + +// parseHiveDescribeRows 解析 Hive DESCRIBE 的输出行，识别普通列与 # Partition Information 分区键列。 +func parseHiveDescribeRows(rows []map[string]interface{}) []connection.ColumnDefinition { + columns := make([]connection.ColumnDefinition, 0, len(rows)) + indexByName := make(map[string]int, len(rows)) + inPartitionSection := false + + for _, row := range rows { + colName := strings.TrimSpace(rowText(row, "col_name", "name", "column_name", "field")) + dataType := strings.TrimSpace(rowText(row, "data_type", "type")) + comment := strings.TrimSpace(rowText(row, "comment", "remarks")) + if comment == "from deserializer" || strings.EqualFold(comment, "null") { + comment = "" + } + + if colName == "" { + continue + } + if strings.HasPrefix(colName, "#") { + lowerHeader := strings.ToLower(colName) + switch { + case strings.Contains(lowerHeader, "partition information"): + inPartitionSection = true + case strings.Contains(lowerHeader, "col_name"): + // 表头行跳过，保持当前段落状态 + default: + inPartitionSection = false + } + continue + } + if dataType == "" { + continue + } + + lowerName := strings.ToLower(colName) + if inPartitionSection { + if idx, exists := indexByName[lowerName]; exists { + columns[idx].Extra = "PARTITION KEY" + if comment != "" && columns[idx].Comment == "" { + columns[idx].Comment = comment + } + } else { + indexByName[lowerName] = len(columns) + columns = append(columns, connection.ColumnDefinition{ + Name: colName, + Type: dataType, + Nullable: "YES", + Comment: comment, + Extra: "PARTITION KEY", + }) + } + continue + } + + if _, exists := indexByName[lowerName]; exists { + continue + } + indexByName[lowerName] = len(columns) + columns = append(columns, connection.ColumnDefinition{ + Name: colName, + Type: dataType, + Nullable: "YES", + Comment: comment, + }) + } + return columns +} + +func (h *HiveDB) GetAllColumns(dbName string) ([]connection.ColumnDefinitionWithTable, error) { + ctx := metadataContextFor(h) + targetDB := h.effectiveDatabase(dbName) + if cols, err := h.fetchAllColumnsViaThrift(ctx, targetDB); err == nil && len(cols) > 0 { + return cols, nil + } + tables, err := h.GetTables(targetDB) + if err != nil { + return nil, err + } + var all []connection.ColumnDefinitionWithTable + for _, table := range tables { + cols, colErr := h.GetColumns(targetDB, table) + if colErr != nil { + continue + } + for _, col := range cols { + all = append(all, connection.ColumnDefinitionWithTable{ + TableName: table, + Name: col.Name, + Type: col.Type, + }) + } + } + return all, nil +} + +func (h *HiveDB) fetchAllColumnsViaThrift(ctx context.Context, dbName string) ([]connection.ColumnDefinitionWithTable, error) { + h.execMu.Lock() + defer h.execMu.Unlock() + if h.transport == nil { + return nil, localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + var op hiveOperationHandle + var status hiveStatus + err := h.transport.call(ctx, "GetColumns", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) // req + w.writeFieldBegin(thriftTypeStruct, 1) // sessionHandle + writeHiveSessionHandle(w, h.sessionID) + w.writeFieldBegin(thriftTypeString, 3) // schemaName + w.writeString(dbName) + w.writeFieldBegin(thriftTypeString, 4) // tableName + w.writeString("%") + w.writeFieldBegin(thriftTypeString, 5) // columnName + w.writeString("%") + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 2 && ft == thriftTypeStruct: + if op, err = readHiveOperationHandle(r); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return nil, err + } + if err := status.toError(); err != nil { + return nil, err + } + defer func() { + cctx, cancel := context.WithTimeout(context.Background(), hiveCancelTimeout) + defer cancel() + _ = h.transport.closeOperation(cctx, op) + }() + cancelOp := func() { + cctx, cancel := context.WithTimeout(context.Background(), hiveCancelTimeout) + defer cancel() + _ = h.transport.cancelOperation(cctx, op) + } + if _, err := h.waitOperationFinished(ctx, op, cancelOp); err != nil { + return nil, err + } + + collector := &hiveRowCollector{} + colsMeta, err := h.transport.getResultSetMetadata(ctx, op) + if err != nil { + return nil, err + } + if err := collector.onColumns(uniqueHiveColumnNames(colsMeta)); err != nil { + return nil, err + } + for { + batch, fetchErr := h.transport.fetchResults(ctx, op, hiveFetchNext, defaultHiveFetchBatchSize) + if fetchErr != nil { + return nil, fetchErr + } + if len(batch) == 0 { + break + } + for _, row := range batch { + if err := collector.onRow(row); err != nil { + return nil, err + } + } + } + result := make([]connection.ColumnDefinitionWithTable, 0, len(collector.data)) + for _, row := range collector.data { + tableName := rowText(row, "TABLE_NAME", "table_name") + colName := rowText(row, "COLUMN_NAME", "column_name") + typeName := rowText(row, "TYPE_NAME", "type_name", "DATA_TYPE") + if tableName != "" && colName != "" { + result = append(result, connection.ColumnDefinitionWithTable{ + TableName: tableName, + Name: colName, + Type: typeName, + }) + } + } + return result, nil +} + +func (h *HiveDB) GetIndexes(string, string) ([]connection.IndexDefinition, error) { + return []connection.IndexDefinition{}, nil +} + +func (h *HiveDB) GetForeignKeys(string, string) ([]connection.ForeignKeyDefinition, error) { + return []connection.ForeignKeyDefinition{}, nil +} + +func (h *HiveDB) GetTriggers(string, string) ([]connection.TriggerDefinition, error) { + return []connection.TriggerDefinition{}, nil +} + +func (h *HiveDB) ApplyChangesContext(ctx context.Context, tableName string, changes connection.ChangeSet) error { + reject := localizedDatabaseRuntimeError("db.backend.error.hive_row_edit_unsupported", nil) + if len(changes.Updates) > 0 || len(changes.Deletes) > 0 { + return reject + } + targetDB, targetTable := resolveHiveTableTarget("", tableName, h.database) + columns, _ := h.GetColumns(targetDB, targetTable) + return applyInsertOnlyChanges(ctx, h.ExecContext, columns, targetTable, changes, insertOnlyDialect{ + quoteTable: func(name string) string { return quoteHiveTable(targetDB, name) }, + quoteIdent: quoteHiveIdent, + escapeBackslash: true, + rowsPerStatement: 200, + }, reject) +} + +func (h *HiveDB) ApplyChanges(tableName string, changes connection.ChangeSet) error { + return h.ApplyChangesContext(context.Background(), tableName, changes) +} + +func firstRowValueText(row map[string]interface{}) string { + for _, value := range row { + if value != nil { + if text := strings.TrimSpace(rowText(map[string]interface{}{"v": value}, "v")); text != "" { + return text + } + } + } + return "" +} diff --git a/internal/db/hive_impl_test.go b/internal/db/hive_impl_test.go new file mode 100644 index 000000000..f30f40f5d --- /dev/null +++ b/internal/db/hive_impl_test.go @@ -0,0 +1,146 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "bytes" + "testing" + + "GoNavi-Wails/internal/connection" +) + +func TestResolveHiveTargetAndURI(t *testing.T) { + defaultTarget := resolveHiveTarget(connection.ConnectionConfig{Type: "hive"}) + if defaultTarget.host != "127.0.0.1" || defaultTarget.port != 10000 || defaultTarget.user != "hive" || defaultTarget.database != "default" { + t.Fatalf("unexpected default target: %+v", defaultTarget) + } + + jdbcTarget := resolveHiveTarget(connection.ConnectionConfig{ + Type: "hive", + URI: "jdbc:hive2://hs2.example.com:10001/dw;auth=noSasl;hive.execution.engine=tez", + }) + if jdbcTarget.host != "hs2.example.com" || jdbcTarget.port != 10001 || jdbcTarget.database != "dw" { + t.Fatalf("unexpected jdbc target: %+v", jdbcTarget) + } + if !isHiveNoSASL(jdbcTarget.params) { + t.Fatalf("expected auth=noSasl to enable noSASL: %+v", jdbcTarget.params) + } + conf := buildHiveSessionConfiguration(jdbcTarget.database, jdbcTarget.params) + if conf["use:database"] != "dw" || conf["set:hiveconf:hive.execution.engine"] != "tez" || conf["set:hiveconf:hive.resultset.use.unique.column.names"] != "false" { + t.Fatalf("unexpected session configuration: %+v", conf) + } + + uriTarget := resolveHiveTarget(connection.ConnectionConfig{ + Type: "hive", + URI: "hive://analyst:secret@10.0.0.8:10000/analytics?ssl=true&hiveconf:mapreduce.job.queuename=prod", + }) + if uriTarget.host != "10.0.0.8" || uriTarget.user != "analyst" || uriTarget.password != "secret" || uriTarget.database != "analytics" || !uriTarget.useSSL { + t.Fatalf("unexpected uri target: %+v", uriTarget) + } +} + +func TestParseHiveVersionAndVariants(t *testing.T) { + cases := map[string]struct { + version string + variant string + }{ + "4.0.1 r3af4517eb8cfd9407ad34ed78a0b48b57dfaa264": {version: "4.0.1", variant: "v4"}, + "3.1.3000.7.1.8.0-801 r123456": {version: "3.1.3000", variant: "v3"}, + } + for banner, want := range cases { + gotVer := parseHiveVersion(banner) + if gotVer != want.version { + t.Fatalf("parseHiveVersion(%q) = %q, want %q", banner, gotVer, want.version) + } + var state driverVariantState + if err := state.resolve("hive", connection.ConnectionConfig{Type: "hive"}, gotVer); err != nil { + t.Fatalf("resolve(%q): %v", gotVer, err) + } + gotVariant, _ := state.DriverVariantInfo() + if gotVariant != want.variant { + t.Fatalf("variant for %q = %q, want %q", banner, gotVariant, want.variant) + } + } +} + +func TestParseHiveDescribeRows(t *testing.T) { + rows := []map[string]interface{}{ + {"col_name": "order_id", "data_type": "string", "comment": "订单号"}, + {"col_name": "qty", "data_type": "int", "comment": "from deserializer"}, + {"col_name": "dt", "data_type": "string", "comment": "分区日期"}, + {"col_name": "", "data_type": nil, "comment": nil}, + {"col_name": "# Partition Information", "data_type": nil, "comment": nil}, + {"col_name": "# col_name", "data_type": "data_type", "comment": "comment"}, + {"col_name": "dt", "data_type": "string", "comment": "分区日期"}, + } + cols := parseHiveDescribeRows(rows) + if len(cols) != 3 { + t.Fatalf("expected 3 columns, got %d: %#v", len(cols), cols) + } + if cols[0].Name != "order_id" || cols[0].Comment != "订单号" || cols[0].Extra != "" { + t.Fatalf("unexpected first column: %#v", cols[0]) + } + if cols[1].Name != "qty" || cols[1].Comment != "" { + t.Fatalf("expected 'from deserializer' comment to be stripped: %#v", cols[1]) + } + if cols[2].Name != "dt" || cols[2].Extra != "PARTITION KEY" { + t.Fatalf("expected dt to be marked as PARTITION KEY: %#v", cols[2]) + } +} + +func TestHiveBindArgsAndNormalizeStatement(t *testing.T) { + if got := normalizeHiveStatement("SELECT * FROM `dw`.`ods_orders`; \n\t"); got != "SELECT * FROM `dw`.`ods_orders`" { + t.Fatalf("unexpected normalized statement: %q", got) + } + bound, err := hiveBindArgs("/* comment ? */ SELECT * FROM `t?` WHERE note = '?' AND id = ? AND active = ? AND tag = ?", []any{42, true, "O'Reilly"}) + if err != nil { + t.Fatalf("hiveBindArgs error: %v", err) + } + want := `/* comment ? */ SELECT * FROM ` + "`t?`" + ` WHERE note = '?' AND id = 42 AND active = TRUE AND tag = 'O\'Reilly'` + if bound != want { + t.Fatalf("hiveBindArgs = %q, want %q", bound, want) + } +} + +func TestHiveThriftColumnarRowSetDecode(t *testing.T) { + w := newHiveWriter() + w.writeFieldBegin(thriftTypeI64, 1) // startRowOffset + w.writeI64(0) + w.writeFieldBegin(thriftTypeList, 3) // columns + w.buf.WriteByte(thriftTypeStruct) + w.writeI32(2) // 2 columns + + // Column 0: i32Val [10, 20], second is NULL + w.writeFieldBegin(thriftTypeStruct, 4) // i32Val + w.writeFieldBegin(thriftTypeList, 1) + w.buf.WriteByte(thriftTypeI32) + w.writeI32(2) + w.writeI32(10) + w.writeI32(20) + w.writeFieldBegin(thriftTypeString, 2) // nulls bitmap: row 1 is null (bit 1 -> 0x02) + w.writeBytes([]byte{0x02}) + w.writeFieldStop() + w.writeFieldStop() + + // Column 1: stringVal ["alpha", "beta"], none NULL + w.writeFieldBegin(thriftTypeStruct, 7) // stringVal + w.writeFieldBegin(thriftTypeList, 1) + w.buf.WriteByte(thriftTypeString) + w.writeI32(2) + w.writeString("alpha") + w.writeString("beta") + w.writeFieldBegin(thriftTypeString, 2) + w.writeBytes([]byte{0x00}) + w.writeFieldStop() + w.writeFieldStop() + + w.writeFieldStop() // end TRowSet + + rows, err := readHiveRowSet(newHiveReader(bytes.NewReader(w.bytes()))) + if err != nil { + t.Fatalf("readHiveRowSet error: %v", err) + } + if len(rows) != 2 || rows[0][0] != int64(10) || rows[0][1] != "alpha" || rows[1][0] != nil || rows[1][1] != "beta" { + t.Fatalf("unexpected decoded rows: %#v", rows) + } +} diff --git a/internal/db/hive_live_test.go b/internal/db/hive_live_test.go new file mode 100644 index 000000000..3585a3c56 --- /dev/null +++ b/internal/db/hive_live_test.go @@ -0,0 +1,187 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + + "GoNavi-Wails/internal/connection" +) + +func TestHiveLiveSmoke(t *testing.T) { + configs := hiveLiveConfigs(t) + for _, tc := range configs { + t.Run(tc.name, func(t *testing.T) { + client := &HiveDB{} + if err := client.Connect(tc.config); err != nil { + t.Fatalf("connect: %v", err) + } + defer client.Close() + + variant, version := client.DriverVariantInfo() + t.Logf("Hive server %s resolved to variant %s", version, variant) + if variant == "" || version == "" { + t.Fatal("variant and version must be resolved from SELECT version()") + } + + if err := client.Ping(); err != nil { + t.Fatalf("ping: %v", err) + } + + databases, err := client.GetDatabases() + if err != nil || !slices.Contains(databases, "default") { + t.Fatalf("databases %v: %v", databases, err) + } + + targetDB := tc.config.Database + if slices.Contains(databases, "dw") { + targetDB = "dw" + } + tables, err := client.GetTables(targetDB) + if err != nil { + t.Fatalf("GetTables(%s): %v", targetDB, err) + } + t.Logf("tables in %s: %v", targetDB, tables) + + if slices.Contains(tables, "ods_orders") { + cols, err := client.GetColumns(targetDB, "ods_orders") + if err != nil || len(cols) == 0 { + t.Fatalf("GetColumns(ods_orders) %#v: %v", cols, err) + } + var foundPartition bool + for _, col := range cols { + if col.Name == "dt" && col.Extra == "PARTITION KEY" { + foundPartition = true + } + } + if !foundPartition { + t.Fatalf("expected dt column to have PARTITION KEY extra, got %#v", cols) + } + + ddl, err := client.GetCreateStatement(targetDB, "ods_orders") + if err != nil || !strings.Contains(strings.ToUpper(ddl), "CREATE") { + t.Fatalf("GetCreateStatement(ods_orders) %q: %v", ddl, err) + } + + rows, colNames, err := client.Query("SELECT order_id, user_id, qty FROM `" + targetDB + "`.`ods_orders` LIMIT 5 OFFSET 0") + if err != nil { + t.Fatalf("Query ods_orders: %v", err) + } + t.Logf("queried %d rows from ods_orders, columns=%v", len(rows), colNames) + if slices.Contains(colNames, "ods_orders.order_id") { + t.Fatalf("expected unqualified column names, got %v", colNames) + } + + allCols, err := client.GetAllColumns(targetDB) + if err != nil || len(allCols) == 0 { + t.Fatalf("GetAllColumns(%s) %v: %v", targetDB, allCols, err) + } + t.Logf("GetAllColumns(%s) returned %d columns", targetDB, len(allCols)) + } + + _, _ = client.Exec("DROP VIEW IF EXISTS `default`.`gonavi_hive_smoke_v`") + _, _ = client.Exec("DROP TABLE IF EXISTS `default`.`gonavi_hive_smoke_t`") + mustExec(t, client, "CREATE TABLE `default`.`gonavi_hive_smoke_t` (`id` INT COMMENT '主键', `name` STRING COMMENT '名称')") + defer func() { _, _ = client.Exec("DROP TABLE IF EXISTS `default`.`gonavi_hive_smoke_t`") }() + + if err := client.ApplyChanges("`default`.`gonavi_hive_smoke_t`", connection.ChangeSet{ + Inserts: []map[string]interface{}{ + {"id": 1, "name": "alice"}, + {"id": 2, "name": "bob"}, + }, + }); err != nil { + t.Fatalf("ApplyChanges insert: %v", err) + } + if err := client.ApplyChanges("`default`.`gonavi_hive_smoke_t`", connection.ChangeSet{ + Deletes: []map[string]interface{}{{"id": 1}}, + }); err == nil { + t.Fatal("expected single-row delete in ApplyChanges to be rejected for Hive") + } + + mustExec(t, client, "CREATE VIEW `default`.`gonavi_hive_smoke_v` AS SELECT `id`, `name` FROM `default`.`gonavi_hive_smoke_t` WHERE `id` >= 1") + defer func() { _, _ = client.Exec("DROP VIEW IF EXISTS `default`.`gonavi_hive_smoke_v`") }() + + defaultTables, err := client.GetTables("default") + if err != nil || !slices.Contains(defaultTables, "gonavi_hive_smoke_t") || slices.Contains(defaultTables, "gonavi_hive_smoke_v") { + t.Fatalf("default tables (must include table and exclude view) %v: %v", defaultTables, err) + } + + viewDDL, err := client.GetCreateStatement("default", "gonavi_hive_smoke_v") + if err != nil || !strings.Contains(strings.ToUpper(viewDDL), "CREATE VIEW") { + t.Fatalf("GetCreateStatement(view) %q: %v", viewDDL, err) + } + + paged, _, err := client.QueryContextWithArgs(t.Context(), "SELECT `name` FROM `default`.`gonavi_hive_smoke_t` WHERE `id` >= ? ORDER BY `id` LIMIT 1 OFFSET 1", []any{1}) + if err != nil || len(paged) != 1 || paged[0]["name"] != "bob" { + t.Fatalf("paged query with args %v: %v", paged, err) + } + }) + } +} + +type namedHiveLiveConfig struct { + name string + config connection.ConnectionConfig +} + +func hiveLiveConfigs(t *testing.T) []namedHiveLiveConfig { + t.Helper() + var list []namedHiveLiveConfig + if rawAddrs := strings.TrimSpace(os.Getenv("GONAVI_HIVE_TEST_ADDRS")); rawAddrs != "" { + for _, addr := range strings.Split(rawAddrs, ",") { + cfg := liveConfig(t, "hive", addr, "hive", "default") + cfg.Timeout = 30 + list = append(list, namedHiveLiveConfig{name: addr, config: cfg}) + } + } + if sshSpec := strings.TrimSpace(os.Getenv("GONAVI_HIVE_TEST_SSH")); sshSpec != "" { + // 格式：user@host:port（例如 lskam@frp-fun.com:49692） + user := "root" + hostPort := sshSpec + if at := strings.Index(sshSpec, "@"); at > 0 { + user = sshSpec[:at] + hostPort = sshSpec[at+1:] + } + sshHost, sshPortText, ok := strings.Cut(hostPort, ":") + sshPort := 22 + if ok { + if p, err := strconv.Atoi(sshPortText); err == nil && p > 0 { + sshPort = p + } + } else { + sshHost = hostPort + } + home, _ := os.UserHomeDir() + keyPath := os.Getenv("GONAVI_HIVE_TEST_SSH_KEY") + if keyPath == "" && home != "" { + keyPath = filepath.Join(home, ".ssh", "id_ed25519") + } + list = append(list, namedHiveLiveConfig{ + name: "ssh-" + sshSpec, + config: connection.ConnectionConfig{ + Type: "hive", + Host: "127.0.0.1", + Port: 10000, + User: "hive", + Database: "dw", + Timeout: 30, + UseSSH: true, + SSH: connection.SSHConfig{ + Host: sshHost, + Port: sshPort, + User: user, + KeyPath: keyPath, + }, + }, + }) + } + if len(list) == 0 { + t.Skip("set GONAVI_HIVE_TEST_ADDRS=host:port or GONAVI_HIVE_TEST_SSH=user@host:port to run Hive live tests") + } + return list +} diff --git a/internal/db/hive_protocol.go b/internal/db/hive_protocol.go new file mode 100644 index 000000000..1d287034d --- /dev/null +++ b/internal/db/hive_protocol.go @@ -0,0 +1,564 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "context" + "fmt" + "strings" +) + +// HiveServer2 TCLIService 协议常量定义。 +const ( + hiveProtocolV10 int32 = 9 // HIVE_CLI_SERVICE_PROTOCOL_V10 (Hive 3.x) + hiveProtocolV11 int32 = 10 // HIVE_CLI_SERVICE_PROTOCOL_V11 (Hive 4.x) + + hiveStatusSuccess int32 = 0 + hiveStatusSuccessWithInfo int32 = 1 + hiveStatusStillExecuting int32 = 2 + hiveStatusError int32 = 3 + hiveStatusInvalidHandle int32 = 4 + + hiveOpInitialized int32 = 0 + hiveOpRunning int32 = 1 + hiveOpFinished int32 = 2 + hiveOpCanceled int32 = 3 + hiveOpClosed int32 = 4 + hiveOpError int32 = 5 + hiveOpUnknown int32 = 6 + hiveOpPending int32 = 7 + hiveOpTimedout int32 = 8 + + hiveFetchNext int32 = 0 + hiveFetchFirst int32 = 4 + + defaultHiveFetchBatchSize int64 = 1000 +) + +type hiveHandleID struct { + guid []byte + secret []byte +} + +type hiveOperationHandle struct { + id hiveHandleID + opType int32 + hasResultSet bool + modifiedRowCount float64 + hasModifiedCount bool +} + +type hiveStatus struct { + statusCode int32 + infoMessages []string + sqlState string + errorCode int32 + errorMessage string +} + +type hiveOperationStatus struct { + status hiveStatus + operationState int32 + sqlState string + errorCode int32 + errorMessage string + hasResultSet bool +} + +// hiveQueryError 表示 HiveServer2 服务端返回的 SQL 执行错误。 +type hiveQueryError struct { + sqlState string + errorCode int32 + message string +} + +func (e *hiveQueryError) Error() string { + state := strings.TrimSpace(e.sqlState) + if state == "" { + state = "HY000" + } + msg := strings.TrimSpace(e.message) + if msg == "" { + msg = "unknown Hive error" + } + return localizedDatabaseRuntimeError("db.backend.error.hive_query_failed", map[string]any{ + "sqlState": state, + "code": e.errorCode, + "message": msg, + }).Error() +} + +func (s hiveStatus) toError() error { + if s.statusCode == hiveStatusSuccess || s.statusCode == hiveStatusSuccessWithInfo || s.statusCode == hiveStatusStillExecuting { + return nil + } + msg := strings.TrimSpace(s.errorMessage) + if msg == "" && len(s.infoMessages) > 0 { + msg = strings.TrimSpace(strings.Join(s.infoMessages, "; ")) + } + if msg == "" { + msg = fmt.Sprintf("Hive status code %d", s.statusCode) + } + return &hiveQueryError{ + sqlState: s.sqlState, + errorCode: s.errorCode, + message: msg, + } +} + +func writeHiveHandleIdentifier(w *hiveWriter, id hiveHandleID) { + w.writeFieldBegin(thriftTypeString, 1) + w.writeBytes(id.guid) + w.writeFieldBegin(thriftTypeString, 2) + w.writeBytes(id.secret) + w.writeFieldStop() +} + +func writeHiveSessionHandle(w *hiveWriter, sessionID hiveHandleID) { + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveHandleIdentifier(w, sessionID) + w.writeFieldStop() +} + +func writeHiveOperationHandle(w *hiveWriter, op hiveOperationHandle) { + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveHandleIdentifier(w, op.id) + w.writeFieldBegin(thriftTypeI32, 2) + w.writeI32(op.opType) + w.writeFieldBegin(thriftTypeBool, 3) + w.writeBool(op.hasResultSet) + if op.hasModifiedCount { + w.writeFieldBegin(thriftTypeDouble, 4) + w.writeDouble(op.modifiedRowCount) + } + w.writeFieldStop() +} + +func readHiveHandleIdentifier(r *hiveReader) (hiveHandleID, error) { + var id hiveHandleID + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return id, err + } + if ft == thriftTypeStop { + return id, nil + } + switch { + case fid == 1 && ft == thriftTypeString: + if id.guid, err = r.readBytes(); err != nil { + return id, err + } + case fid == 2 && ft == thriftTypeString: + if id.secret, err = r.readBytes(); err != nil { + return id, err + } + default: + if err := r.skip(ft); err != nil { + return id, err + } + } + } +} + +func readHiveSessionHandle(r *hiveReader) (hiveHandleID, error) { + var id hiveHandleID + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return id, err + } + if ft == thriftTypeStop { + return id, nil + } + if fid == 1 && ft == thriftTypeStruct { + if id, err = readHiveHandleIdentifier(r); err != nil { + return id, err + } + } else if err := r.skip(ft); err != nil { + return id, err + } + } +} + +func readHiveOperationHandle(r *hiveReader) (hiveOperationHandle, error) { + var op hiveOperationHandle + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return op, err + } + if ft == thriftTypeStop { + return op, nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if op.id, err = readHiveHandleIdentifier(r); err != nil { + return op, err + } + case fid == 2 && ft == thriftTypeI32: + if op.opType, err = r.readI32(); err != nil { + return op, err + } + case fid == 3 && ft == thriftTypeBool: + if op.hasResultSet, err = r.readBool(); err != nil { + return op, err + } + case fid == 4 && ft == thriftTypeDouble: + if op.modifiedRowCount, err = r.readDouble(); err != nil { + return op, err + } + op.hasModifiedCount = true + default: + if err := r.skip(ft); err != nil { + return op, err + } + } + } +} + +func readHiveStatus(r *hiveReader) (hiveStatus, error) { + var st hiveStatus + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return st, err + } + if ft == thriftTypeStop { + return st, nil + } + switch { + case fid == 1 && ft == thriftTypeI32: + if st.statusCode, err = r.readI32(); err != nil { + return st, err + } + case fid == 2 && ft == thriftTypeList: + et, size, err := r.readListBegin() + if err != nil { + return st, err + } + for i := 0; i < size; i++ { + if et == thriftTypeString { + s, err := r.readString() + if err != nil { + return st, err + } + st.infoMessages = append(st.infoMessages, s) + } else if err := r.skip(et); err != nil { + return st, err + } + } + case fid == 3 && ft == thriftTypeString: + if st.sqlState, err = r.readString(); err != nil { + return st, err + } + case fid == 4 && ft == thriftTypeI32: + if st.errorCode, err = r.readI32(); err != nil { + return st, err + } + case fid == 5 && ft == thriftTypeString: + if st.errorMessage, err = r.readString(); err != nil { + return st, err + } + default: + if err := r.skip(ft); err != nil { + return st, err + } + } + } +} + +func readSingleStructField(r *hiveReader, decode func(*hiveReader) error) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + if fid == 0 && ft == thriftTypeStruct { + if err := decode(r); err != nil { + return err + } + } else if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } +} + +// openSession 调用 TCLIService.OpenSession 打开会话并协商协议版本。 +func (t *hiveTransport) openSession(ctx context.Context, clientProtocol int32, username, password string, conf map[string]string) (hiveHandleID, int32, error) { + var sessionID hiveHandleID + var serverProtocol int32 + var status hiveStatus + err := t.call(ctx, "OpenSession", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) // req + w.writeFieldBegin(thriftTypeI32, 1) // client_protocol + w.writeI32(clientProtocol) + if username != "" { + w.writeFieldBegin(thriftTypeString, 2) + w.writeString(username) + } + if password != "" { + w.writeFieldBegin(thriftTypeString, 3) + w.writeString(password) + } + if len(conf) > 0 { + w.writeFieldBegin(thriftTypeMap, 4) + w.writeStringMap(conf) + } + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 2 && ft == thriftTypeI32: + if serverProtocol, err = r.readI32(); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 3 && ft == thriftTypeStruct: + if sessionID, err = readHiveSessionHandle(r); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return sessionID, 0, err + } + if err := status.toError(); err != nil { + return sessionID, 0, err + } + return sessionID, serverProtocol, nil +} + +func (t *hiveTransport) closeSession(ctx context.Context, sessionID hiveHandleID) error { + var status hiveStatus + err := t.call(ctx, "CloseSession", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) // req + w.writeFieldBegin(thriftTypeStruct, 1) // sessionHandle + writeHiveSessionHandle(w, sessionID) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + if fid == 1 && ft == thriftTypeStruct { + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + } else if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + }) + }) + if err != nil { + return err + } + return status.toError() +} + +func (t *hiveTransport) executeStatement(ctx context.Context, sessionID hiveHandleID, statement string, confOverlay map[string]string, runAsync bool, queryTimeoutSec int64) (hiveOperationHandle, error) { + var op hiveOperationHandle + var status hiveStatus + err := t.call(ctx, "ExecuteStatement", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) // req + w.writeFieldBegin(thriftTypeStruct, 1) // sessionHandle + writeHiveSessionHandle(w, sessionID) + w.writeFieldBegin(thriftTypeString, 2) // statement + w.writeString(statement) + if len(confOverlay) > 0 { + w.writeFieldBegin(thriftTypeMap, 3) + w.writeStringMap(confOverlay) + } + w.writeFieldBegin(thriftTypeBool, 4) // runAsync + w.writeBool(runAsync) + if queryTimeoutSec > 0 { + w.writeFieldBegin(thriftTypeI64, 5) // queryTimeout + w.writeI64(queryTimeoutSec) + } + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 2 && ft == thriftTypeStruct: + if op, err = readHiveOperationHandle(r); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return op, err + } + if err := status.toError(); err != nil { + return op, err + } + return op, nil +} + +func (t *hiveTransport) getOperationStatus(ctx context.Context, op hiveOperationHandle) (hiveOperationStatus, error) { + var resp hiveOperationStatus + err := t.call(ctx, "GetOperationStatus", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveOperationHandle(w, op) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if resp.status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 2 && ft == thriftTypeI32: + if resp.operationState, err = r.readI32(); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 3 && ft == thriftTypeString: + if resp.sqlState, err = r.readString(); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 4 && ft == thriftTypeI32: + if resp.errorCode, err = r.readI32(); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 5 && ft == thriftTypeString: + if resp.errorMessage, err = r.readString(); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 9 && ft == thriftTypeBool: + if resp.hasResultSet, err = r.readBool(); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return resp, err + } + if err := resp.status.toError(); err != nil { + return resp, err + } + return resp, nil +} + +func (t *hiveTransport) cancelOperation(ctx context.Context, op hiveOperationHandle) error { + var status hiveStatus + err := t.call(ctx, "CancelOperation", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveOperationHandle(w, op) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + if fid == 1 && ft == thriftTypeStruct { + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + } else if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + }) + }) + if err != nil { + return err + } + return status.toError() +} + +func (t *hiveTransport) closeOperation(ctx context.Context, op hiveOperationHandle) error { + var status hiveStatus + err := t.call(ctx, "CloseOperation", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveOperationHandle(w, op) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + if fid == 1 && ft == thriftTypeStruct { + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + } else if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + }) + }) + if err != nil { + return err + } + return status.toError() +} diff --git a/internal/db/hive_rows.go b/internal/db/hive_rows.go new file mode 100644 index 000000000..1df15b906 --- /dev/null +++ b/internal/db/hive_rows.go @@ -0,0 +1,561 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "context" + "fmt" + "strings" +) + +type hiveColumnMeta struct { + name string + typeName string + position int32 + comment string +} + +type hiveRowSink interface { + onColumns(names []string) error + onRow(values []any) error +} + +func (t *hiveTransport) getResultSetMetadata(ctx context.Context, op hiveOperationHandle) ([]hiveColumnMeta, error) { + var cols []hiveColumnMeta + var status hiveStatus + err := t.call(ctx, "GetResultSetMetadata", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) + w.writeFieldBegin(thriftTypeStruct, 1) + writeHiveOperationHandle(w, op) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 2 && ft == thriftTypeStruct: + if cols, err = readHiveTableSchema(r); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return nil, err + } + if err := status.toError(); err != nil { + return nil, err + } + return cols, nil +} + +func (t *hiveTransport) fetchResults(ctx context.Context, op hiveOperationHandle, orientation int32, maxRows int64) ([][]any, error) { + var rows [][]any + var status hiveStatus + err := t.call(ctx, "FetchResults", func(w *hiveWriter) { + w.writeFieldBegin(thriftTypeStruct, 1) // req + w.writeFieldBegin(thriftTypeStruct, 1) // operationHandle + writeHiveOperationHandle(w, op) + w.writeFieldBegin(thriftTypeI32, 2) // orientation + w.writeI32(orientation) + w.writeFieldBegin(thriftTypeI64, 3) // maxRows + w.writeI64(maxRows) + w.writeFieldBegin(thriftTypeI16, 4) // fetchType = 0 (query output) + w.writeI16(0) + w.writeFieldStop() + }, func(r *hiveReader) error { + return readSingleStructField(r, func(r *hiveReader) error { + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return wrapHiveProtocolError(err) + } + if ft == thriftTypeStop { + return nil + } + switch { + case fid == 1 && ft == thriftTypeStruct: + if status, err = readHiveStatus(r); err != nil { + return wrapHiveProtocolError(err) + } + case fid == 3 && ft == thriftTypeStruct: + if rows, err = readHiveRowSet(r); err != nil { + return wrapHiveProtocolError(err) + } + default: + if err := r.skip(ft); err != nil { + return wrapHiveProtocolError(err) + } + } + } + }) + }) + if err != nil { + return nil, err + } + if err := status.toError(); err != nil { + return nil, err + } + return rows, nil +} + +func readHiveTableSchema(r *hiveReader) ([]hiveColumnMeta, error) { + var cols []hiveColumnMeta + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + return cols, nil + } + if fid == 1 && ft == thriftTypeList { + et, size, err := r.readListBegin() + if err != nil { + return nil, err + } + cols = make([]hiveColumnMeta, 0, size) + for i := 0; i < size; i++ { + if et != thriftTypeStruct { + if err := r.skip(et); err != nil { + return nil, err + } + continue + } + col, err := readHiveColumnDesc(r) + if err != nil { + return nil, err + } + cols = append(cols, col) + } + } else if err := r.skip(ft); err != nil { + return nil, err + } + } +} + +func readHiveColumnDesc(r *hiveReader) (hiveColumnMeta, error) { + var col hiveColumnMeta + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return col, err + } + if ft == thriftTypeStop { + return col, nil + } + switch { + case fid == 1 && ft == thriftTypeString: + if col.name, err = r.readString(); err != nil { + return col, err + } + case fid == 2 && ft == thriftTypeStruct: + if col.typeName, err = readHiveTypeDesc(r); err != nil { + return col, err + } + case fid == 3 && ft == thriftTypeI32: + if col.position, err = r.readI32(); err != nil { + return col, err + } + case fid == 4 && ft == thriftTypeString: + if col.comment, err = r.readString(); err != nil { + return col, err + } + default: + if err := r.skip(ft); err != nil { + return col, err + } + } + } +} + +func readHiveTypeDesc(r *hiveReader) (string, error) { + var typeName string + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return "", err + } + if ft == thriftTypeStop { + return typeName, nil + } + if fid == 1 && ft == thriftTypeList { + et, size, err := r.readListBegin() + if err != nil { + return "", err + } + for i := 0; i < size; i++ { + if et != thriftTypeStruct { + if err := r.skip(et); err != nil { + return "", err + } + continue + } + name, err := readHiveTypeEntry(r) + if err != nil { + return "", err + } + if i == 0 { + typeName = name + } + } + } else if err := r.skip(ft); err != nil { + return "", err + } + } +} + +func readHiveTypeEntry(r *hiveReader) (string, error) { + var typeName string + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return "", err + } + if ft == thriftTypeStop { + return typeName, nil + } + if fid == 1 && ft == thriftTypeStruct { + for { + pft, pfid, err := r.readFieldBegin() + if err != nil { + return "", err + } + if pft == thriftTypeStop { + break + } + if pfid == 1 && pft == thriftTypeI32 { + typeID, err := r.readI32() + if err != nil { + return "", err + } + typeName = hiveTypeIDName(typeID) + } else if err := r.skip(pft); err != nil { + return "", err + } + } + } else if err := r.skip(ft); err != nil { + return "", err + } + } +} + +func hiveTypeIDName(id int32) string { + switch id { + case 0: + return "BOOLEAN" + case 1: + return "TINYINT" + case 2: + return "SMALLINT" + case 3: + return "INT" + case 4: + return "BIGINT" + case 5: + return "FLOAT" + case 6: + return "DOUBLE" + case 7: + return "STRING" + case 8: + return "TIMESTAMP" + case 9: + return "BINARY" + case 10: + return "ARRAY" + case 11: + return "MAP" + case 12: + return "STRUCT" + case 13: + return "UNIONTYPE" + case 15: + return "DECIMAL" + case 17: + return "DATE" + case 18: + return "VARCHAR" + case 19: + return "CHAR" + case 20: + return "INTERVAL_YEAR_MONTH" + case 21: + return "INTERVAL_DAY_TIME" + case 22: + return "TIMESTAMPLOCALTZ" + default: + return "STRING" + } +} + +func readHiveRowSet(r *hiveReader) ([][]any, error) { + var legacyRows [][]any + var columns [][]any + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + break + } + switch { + case fid == 2 && ft == thriftTypeList: + et, size, err := r.readListBegin() + if err != nil { + return nil, err + } + for i := 0; i < size; i++ { + if et == thriftTypeStruct { + row, err := readHiveLegacyRow(r) + if err != nil { + return nil, err + } + legacyRows = append(legacyRows, row) + } else if err := r.skip(et); err != nil { + return nil, err + } + } + case fid == 3 && ft == thriftTypeList: + et, size, err := r.readListBegin() + if err != nil { + return nil, err + } + columns = make([][]any, 0, size) + for i := 0; i < size; i++ { + if et == thriftTypeStruct { + colVals, err := readHiveColumn(r) + if err != nil { + return nil, err + } + columns = append(columns, colVals) + } else if err := r.skip(et); err != nil { + return nil, err + } + } + default: + if err := r.skip(ft); err != nil { + return nil, err + } + } + } + if len(columns) > 0 { + rowCount := len(columns[0]) + rows := make([][]any, rowCount) + for rowIdx := 0; rowIdx < rowCount; rowIdx++ { + row := make([]any, len(columns)) + for colIdx, col := range columns { + if rowIdx < len(col) { + row[colIdx] = col[rowIdx] + } + } + rows[rowIdx] = row + } + return rows, nil + } + return legacyRows, nil +} + +func readHiveColumn(r *hiveReader) ([]any, error) { + var values []any + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + return values, nil + } + if ft != thriftTypeStruct || fid < 1 || fid > 8 { + if err := r.skip(ft); err != nil { + return nil, err + } + continue + } + vals, nulls, err := readHiveTypedColumn(r, fid) + if err != nil { + return nil, err + } + for i := range vals { + if isHiveNull(nulls, i) { + vals[i] = nil + } + } + values = vals + } +} + +func readHiveTypedColumn(r *hiveReader, kind int16) ([]any, []byte, error) { + var vals []any + var nulls []byte + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, nil, err + } + if ft == thriftTypeStop { + return vals, nulls, nil + } + switch { + case fid == 1 && ft == thriftTypeList: + _, size, err := r.readListBegin() + if err != nil { + return nil, nil, err + } + vals = make([]any, size) + for i := 0; i < size; i++ { + val, err := readHiveColumnElement(r, kind) + if err != nil { + return nil, nil, err + } + vals[i] = val + } + case fid == 2 && ft == thriftTypeString: + if nulls, err = r.readBytes(); err != nil { + return nil, nil, err + } + default: + if err := r.skip(ft); err != nil { + return nil, nil, err + } + } + } +} + +func readHiveColumnElement(r *hiveReader, kind int16) (any, error) { + switch kind { + case 1: + return r.readBool() + case 2: + b, err := r.readByte() + return int64(int8(b)), err + case 3: + v, err := r.readI16() + return int64(v), err + case 4: + v, err := r.readI32() + return int64(v), err + case 5: + return r.readI64() + case 6: + return r.readDouble() + case 7: + return r.readString() + case 8: + b, err := r.readBytes() + return string(b), err + default: + return nil, fmt.Errorf("unsupported Hive column kind %d", kind) + } +} + +func isHiveNull(nulls []byte, rowIdx int) bool { + byteIdx := rowIdx / 8 + if byteIdx >= len(nulls) { + return false + } + return (nulls[byteIdx] & (1 << (uint(rowIdx) % 8))) != 0 +} + +func readHiveLegacyRow(r *hiveReader) ([]any, error) { + var row []any + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + return row, nil + } + if fid == 1 && ft == thriftTypeList { + et, size, err := r.readListBegin() + if err != nil { + return nil, err + } + row = make([]any, 0, size) + for i := 0; i < size; i++ { + if et == thriftTypeStruct { + val, err := readHiveLegacyColumnValue(r) + if err != nil { + return nil, err + } + row = append(row, val) + } else if err := r.skip(et); err != nil { + return nil, err + } + } + } else if err := r.skip(ft); err != nil { + return nil, err + } + } +} + +func readHiveLegacyColumnValue(r *hiveReader) (any, error) { + var result any + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + return result, nil + } + if ft != thriftTypeStruct { + if err := r.skip(ft); err != nil { + return nil, err + } + continue + } + for { + vft, vfid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if vft == thriftTypeStop { + break + } + if vfid == 1 { + result, err = readHiveColumnElement(r, fid) + if err != nil { + return nil, err + } + } else if err := r.skip(vft); err != nil { + return nil, err + } + } + } +} + +func uniqueHiveColumnNames(cols []hiveColumnMeta) []string { + names := make([]string, len(cols)) + used := make(map[string]int, len(cols)) + for i, col := range cols { + name := strings.TrimSpace(col.name) + if name == "" { + name = fmt.Sprintf("_c%d", i) + } + count := used[name] + used[name] = count + 1 + if count > 0 { + name = fmt.Sprintf("%s_%d", name, count+1) + } + names[i] = name + } + return names +} diff --git a/internal/db/hive_thrift.go b/internal/db/hive_thrift.go new file mode 100644 index 000000000..5db258b1f --- /dev/null +++ b/internal/db/hive_thrift.go @@ -0,0 +1,598 @@ +//go:build gonavi_full_drivers || gonavi_hive_driver + +package db + +import ( + "bufio" + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "math" + "net" + "strings" + "sync" + "sync/atomic" + "time" +) + +// Thrift Binary Protocol 常量定义（对应 Apache Thrift TBinaryProtocol 与 TSaslTransport）。 +const ( + thriftVersionMask uint32 = 0xffff0000 + thriftVersion1 uint32 = 0x80010000 + + thriftMsgCall byte = 1 + thriftMsgReply byte = 2 + thriftMsgException byte = 3 + + thriftTypeStop byte = 0 + thriftTypeVoid byte = 1 + thriftTypeBool byte = 2 + thriftTypeByte byte = 3 + thriftTypeDouble byte = 4 + thriftTypeI16 byte = 6 + thriftTypeI32 byte = 8 + thriftTypeI64 byte = 10 + thriftTypeString byte = 11 + thriftTypeStruct byte = 12 + thriftTypeMap byte = 13 + thriftTypeSet byte = 14 + thriftTypeList byte = 15 + + saslStatusStart byte = 1 + saslStatusOK byte = 2 + saslStatusBad byte = 3 + saslStatusError byte = 4 + saslStatusComplete byte = 5 + + maxThriftFrameBytes = 256 * 1024 * 1024 + maxThriftItemCount = 16 * 1024 * 1024 +) + +// hiveTransport 封装 HiveServer2 的底层 TCP/TLS 连接，支持 SASL PLAIN 帧传输与 NOSASL 原始传输。 +type hiveTransport struct { + mu sync.Mutex + conn net.Conn + reader *bufio.Reader + framed bool + seqID atomic.Int32 +} + +func newHiveTransport(conn net.Conn, framed bool) *hiveTransport { + return &hiveTransport{ + conn: conn, + reader: bufio.NewReaderSize(conn, 64*1024), + framed: framed, + } +} + +func (t *hiveTransport) Close() error { + t.mu.Lock() + defer t.mu.Unlock() + if t.conn == nil { + return nil + } + err := t.conn.Close() + t.conn = nil + return err +} + +// handshakeSASLPlain 执行 TSaslClientTransport 的 PLAIN 握手（适用 NONE / PLAIN / LDAP 认证）。 +func (t *hiveTransport) handshakeSASLPlain(ctx context.Context, username, password string) error { + t.mu.Lock() + defer t.mu.Unlock() + if t.conn == nil { + return localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + stopWatch := watchContextDeadline(ctx, t.conn) + defer stopWatch() + + user := strings.TrimSpace(username) + if user == "" { + user = "anonymous" + } + pass := password + if pass == "" { + pass = "anonymous" + } + + var buf bytes.Buffer + writeSASLMessage(&buf, saslStatusStart, []byte("PLAIN")) + authPayload := make([]byte, 0, len(user)+len(pass)+2) + authPayload = append(authPayload, 0) + authPayload = append(authPayload, user...) + authPayload = append(authPayload, 0) + authPayload = append(authPayload, pass...) + writeSASLMessage(&buf, saslStatusOK, authPayload) + + if _, err := t.conn.Write(buf.Bytes()); err != nil { + return wrapHiveProtocolError(err) + } + + for { + status, payload, err := readSASLMessage(t.reader) + if err != nil { + return wrapHiveProtocolError(err) + } + switch status { + case saslStatusComplete: + return nil + case saslStatusOK: + // Challenge-response 后续轮次对 PLAIN 为空响应 + var next bytes.Buffer + writeSASLMessage(&next, saslStatusOK, nil) + if _, err := t.conn.Write(next.Bytes()); err != nil { + return wrapHiveProtocolError(err) + } + case saslStatusBad, saslStatusError: + detail := strings.TrimSpace(string(payload)) + if detail == "" { + detail = fmt.Sprintf("SASL status %d", status) + } + return localizedDatabaseRuntimeError("db.backend.error.hive_auth_failed", map[string]any{"detail": detail}) + default: + return wrapHiveProtocolError(fmt.Errorf("unexpected SASL negotiation status %d", status)) + } + } +} + +func writeSASLMessage(buf *bytes.Buffer, status byte, payload []byte) { + buf.WriteByte(status) + var lenBuf [4]byte + binary.BigEndian.PutUint32(lenBuf[:], uint32(len(payload))) + buf.Write(lenBuf[:]) + buf.Write(payload) +} + +func readSASLMessage(r io.Reader) (byte, []byte, error) { + var header [5]byte + if _, err := io.ReadFull(r, header[:]); err != nil { + return 0, nil, err + } + status := header[0] + length := binary.BigEndian.Uint32(header[1:5]) + if length > maxThriftFrameBytes { + return 0, nil, fmt.Errorf("SASL frame size %d exceeds limit", length) + } + payload := make([]byte, length) + if length > 0 { + if _, err := io.ReadFull(r, payload); err != nil { + return 0, nil, err + } + } + return status, payload, nil +} + +// call 发送一次 Thrift RPC 请求并读取响应结构。 +func (t *hiveTransport) call(ctx context.Context, method string, encodeArgs func(*hiveWriter), decodeResult func(*hiveReader) error) error { + if err := ctx.Err(); err != nil { + return err + } + t.mu.Lock() + defer t.mu.Unlock() + if t.conn == nil { + return localizedDatabaseRuntimeError("db.backend.error.connection_not_open", nil) + } + stopWatch := watchContextDeadline(ctx, t.conn) + defer stopWatch() + + seq := t.seqID.Add(1) + w := newHiveWriter() + w.writeMessageBegin(method, thriftMsgCall, seq) + encodeArgs(w) + w.writeFieldStop() + + payload := w.bytes() + if t.framed { + var frameHdr [4]byte + binary.BigEndian.PutUint32(frameHdr[:], uint32(len(payload))) + if _, err := t.conn.Write(frameHdr[:]); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + } + if _, err := t.conn.Write(payload); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return err + } + + var r *hiveReader + if t.framed { + var frameHdr [4]byte + if _, err := io.ReadFull(t.reader, frameHdr[:]); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return wrapHiveProtocolError(err) + } + frameLen := binary.BigEndian.Uint32(frameHdr[:]) + if frameLen > maxThriftFrameBytes { + return wrapHiveProtocolError(fmt.Errorf("frame length %d exceeds limit", frameLen)) + } + frame := make([]byte, frameLen) + if _, err := io.ReadFull(t.reader, frame); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return wrapHiveProtocolError(err) + } + r = newHiveReader(bytes.NewReader(frame)) + } else { + r = newHiveReader(t.reader) + } + + _, msgType, _, err := r.readMessageBegin() + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + return wrapHiveProtocolError(err) + } + if msgType == thriftMsgException { + appErr, exErr := r.readApplicationException() + if exErr != nil { + return wrapHiveProtocolError(exErr) + } + return appErr + } + if msgType != thriftMsgReply { + return wrapHiveProtocolError(fmt.Errorf("unexpected Thrift message type %d", msgType)) + } + return decodeResult(r) +} + +func watchContextDeadline(ctx context.Context, conn net.Conn) func() { + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } else { + _ = conn.SetDeadline(time.Time{}) + } + done := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + _ = conn.SetDeadline(time.Now()) + case <-done: + } + }() + return func() { + close(done) + _ = conn.SetDeadline(time.Time{}) + } +} + +func wrapHiveProtocolError(err error) error { + if err == nil { + return nil + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return err + } + return localizedDatabaseRuntimeError("db.backend.error.hive_protocol_error", map[string]any{"detail": err.Error()}) +} + +// hiveWriter 编码 Thrift Binary 协议字节流。 +type hiveWriter struct { + buf bytes.Buffer +} + +func newHiveWriter() *hiveWriter { + return &hiveWriter{} +} + +func (w *hiveWriter) bytes() []byte { + return w.buf.Bytes() +} + +func (w *hiveWriter) writeMessageBegin(name string, msgType byte, seqID int32) { + w.writeI32(int32(thriftVersion1 | uint32(msgType))) + w.writeString(name) + w.writeI32(seqID) +} + +func (w *hiveWriter) writeFieldBegin(fieldType byte, fieldID int16) { + w.buf.WriteByte(fieldType) + w.writeI16(fieldID) +} + +func (w *hiveWriter) writeFieldStop() { + w.buf.WriteByte(thriftTypeStop) +} + +func (w *hiveWriter) writeBool(v bool) { + if v { + w.buf.WriteByte(1) + } else { + w.buf.WriteByte(0) + } +} + +func (w *hiveWriter) writeI16(v int16) { + var b [2]byte + binary.BigEndian.PutUint16(b[:], uint16(v)) + w.buf.Write(b[:]) +} + +func (w *hiveWriter) writeI32(v int32) { + var b [4]byte + binary.BigEndian.PutUint32(b[:], uint32(v)) + w.buf.Write(b[:]) +} + +func (w *hiveWriter) writeI64(v int64) { + var b [8]byte + binary.BigEndian.PutUint64(b[:], uint64(v)) + w.buf.Write(b[:]) +} + +func (w *hiveWriter) writeDouble(v float64) { + w.writeI64(int64(math.Float64bits(v))) +} + +func (w *hiveWriter) writeString(s string) { + w.writeI32(int32(len(s))) + w.buf.WriteString(s) +} + +func (w *hiveWriter) writeBytes(b []byte) { + w.writeI32(int32(len(b))) + w.buf.Write(b) +} + +func (w *hiveWriter) writeStringMap(m map[string]string) { + w.buf.WriteByte(thriftTypeString) + w.buf.WriteByte(thriftTypeString) + w.writeI32(int32(len(m))) + for k, v := range m { + w.writeString(k) + w.writeString(v) + } +} + +// hiveReader 解码 Thrift Binary 协议字节流。 +type hiveReader struct { + r io.Reader +} + +func newHiveReader(r io.Reader) *hiveReader { + return &hiveReader{r: r} +} + +func (r *hiveReader) readMessageBegin() (string, byte, int32, error) { + sizeAndVersion, err := r.readI32() + if err != nil { + return "", 0, 0, err + } + u := uint32(sizeAndVersion) + if (u & thriftVersionMask) != thriftVersion1 { + return "", 0, 0, fmt.Errorf("unsupported Thrift binary version 0x%x", u) + } + msgType := byte(u & 0x000000ff) + name, err := r.readString() + if err != nil { + return "", 0, 0, err + } + seqID, err := r.readI32() + if err != nil { + return "", 0, 0, err + } + return name, msgType, seqID, nil +} + +func (r *hiveReader) readFieldBegin() (byte, int16, error) { + fieldType, err := r.readByte() + if err != nil { + return 0, 0, err + } + if fieldType == thriftTypeStop { + return thriftTypeStop, 0, nil + } + fieldID, err := r.readI16() + if err != nil { + return 0, 0, err + } + return fieldType, fieldID, nil +} + +func (r *hiveReader) readByte() (byte, error) { + var b [1]byte + if _, err := io.ReadFull(r.r, b[:]); err != nil { + return 0, err + } + return b[0], nil +} + +func (r *hiveReader) readBool() (bool, error) { + b, err := r.readByte() + return b != 0, err +} + +func (r *hiveReader) readI16() (int16, error) { + var b [2]byte + if _, err := io.ReadFull(r.r, b[:]); err != nil { + return 0, err + } + return int16(binary.BigEndian.Uint16(b[:])), nil +} + +func (r *hiveReader) readI32() (int32, error) { + var b [4]byte + if _, err := io.ReadFull(r.r, b[:]); err != nil { + return 0, err + } + return int32(binary.BigEndian.Uint32(b[:])), nil +} + +func (r *hiveReader) readI64() (int64, error) { + var b [8]byte + if _, err := io.ReadFull(r.r, b[:]); err != nil { + return 0, err + } + return int64(binary.BigEndian.Uint64(b[:])), nil +} + +func (r *hiveReader) readDouble() (float64, error) { + bits, err := r.readI64() + if err != nil { + return 0, err + } + return math.Float64frombits(uint64(bits)), nil +} + +func (r *hiveReader) readBytes() ([]byte, error) { + size, err := r.readI32() + if err != nil { + return nil, err + } + if size < 0 || size > maxThriftFrameBytes { + return nil, fmt.Errorf("invalid Thrift binary length %d", size) + } + if size == 0 { + return []byte{}, nil + } + buf := make([]byte, size) + if _, err := io.ReadFull(r.r, buf); err != nil { + return nil, err + } + return buf, nil +} + +func (r *hiveReader) readString() (string, error) { + b, err := r.readBytes() + if err != nil { + return "", err + } + return string(b), nil +} + +func (r *hiveReader) readListBegin() (byte, int, error) { + elemType, err := r.readByte() + if err != nil { + return 0, 0, err + } + size, err := r.readI32() + if err != nil { + return 0, 0, err + } + if size < 0 || size > maxThriftItemCount { + return 0, 0, fmt.Errorf("invalid Thrift list size %d", size) + } + return elemType, int(size), nil +} + +func (r *hiveReader) readMapBegin() (byte, byte, int, error) { + keyType, err := r.readByte() + if err != nil { + return 0, 0, 0, err + } + valType, err := r.readByte() + if err != nil { + return 0, 0, 0, err + } + size, err := r.readI32() + if err != nil { + return 0, 0, 0, err + } + if size < 0 || size > maxThriftItemCount { + return 0, 0, 0, fmt.Errorf("invalid Thrift map size %d", size) + } + return keyType, valType, int(size), nil +} + +func (r *hiveReader) skip(fieldType byte) error { + switch fieldType { + case thriftTypeBool, thriftTypeByte: + _, err := r.readByte() + return err + case thriftTypeI16: + _, err := r.readI16() + return err + case thriftTypeI32: + _, err := r.readI32() + return err + case thriftTypeI64, thriftTypeDouble: + _, err := r.readI64() + return err + case thriftTypeString: + _, err := r.readBytes() + return err + case thriftTypeStruct: + for { + ft, _, err := r.readFieldBegin() + if err != nil { + return err + } + if ft == thriftTypeStop { + return nil + } + if err := r.skip(ft); err != nil { + return err + } + } + case thriftTypeMap: + kt, vt, size, err := r.readMapBegin() + if err != nil { + return err + } + for i := 0; i < size; i++ { + if err := r.skip(kt); err != nil { + return err + } + if err := r.skip(vt); err != nil { + return err + } + } + return nil + case thriftTypeList, thriftTypeSet: + et, size, err := r.readListBegin() + if err != nil { + return err + } + for i := 0; i < size; i++ { + if err := r.skip(et); err != nil { + return err + } + } + return nil + default: + return fmt.Errorf("unknown Thrift field type %d", fieldType) + } +} + +func (r *hiveReader) readApplicationException() (error, error) { + var msg string + var exType int32 + for { + ft, fid, err := r.readFieldBegin() + if err != nil { + return nil, err + } + if ft == thriftTypeStop { + break + } + switch { + case fid == 1 && ft == thriftTypeString: + if msg, err = r.readString(); err != nil { + return nil, err + } + case fid == 2 && ft == thriftTypeI32: + if exType, err = r.readI32(); err != nil { + return nil, err + } + default: + if err := r.skip(ft); err != nil { + return nil, err + } + } + } + if strings.TrimSpace(msg) == "" { + msg = fmt.Sprintf("Thrift application exception (type %d)", exType) + } + return errors.New(msg), nil +} diff --git a/internal/db/registry_rows.go b/internal/db/registry_rows.go index b122e12f1..7473e61e1 100644 --- a/internal/db/registry_rows.go +++ b/internal/db/registry_rows.go @@ -1,4 +1,4 @@ -//go:build gonavi_full_drivers || gonavi_cockroachdb_driver || gonavi_kwdb_driver || gonavi_questdb_driver || gonavi_greptimedb_driver || gonavi_timescaledb_driver || gonavi_presto_driver +//go:build gonavi_full_drivers || gonavi_cockroachdb_driver || gonavi_kwdb_driver || gonavi_questdb_driver || gonavi_greptimedb_driver || gonavi_timescaledb_driver || gonavi_presto_driver || gonavi_hive_driver package db diff --git a/shared/i18n/catalog.zip b/shared/i18n/catalog.zip index 89b8e45da..3a3c9e794 100644 Binary files a/shared/i18n/catalog.zip and b/shared/i18n/catalog.zip differ diff --git a/shared/i18n/de-DE.json b/shared/i18n/de-DE.json index feb5a0578..444c9be2d 100644 --- a/shared/i18n/de-DE.json +++ b/shared/i18n/de-DE.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "Sonstige", "plugin.market.anonymous_usage": "Öffentliche Plugins ohne Anmeldung durchsuchen und installieren", "plugin.market.ratings_on_website": "Bewertungen und Kontoverwaltung erfolgen auf der Marktplatz-Website. GoNavi zeigt öffentliche Bewertungen nur an.", - "plugin.market.open_website": "Marktplatz-Website öffnen" + "plugin.market.open_website": "Marktplatz-Website öffnen", + "datasource.variant.hive.v3": "Hive 3.x: Spaltenbasiertes HiveServer2-V10-Ergebnisprotokoll mit LIMIT/OFFSET-Seitenumbruch und Metadaten für externe Tabellen", + "datasource.variant.hive.v4": "Hive 4.x und neuer: Spaltenbasiertes HiveServer2-V11-Ergebnisprotokoll mit Iceberg-/Transaktionstabellen und vollständigen DDL-Metadaten", + "connection_modal.step1.hint.hive": "Data Warehouse / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "HiveServer2-Authentifizierung fehlgeschlagen: {{detail}}", + "db.backend.error.hive_query_failed": "Hive-Abfrage fehlgeschlagen (SQLState {{sqlState}}, Fehlercode {{code}}): {{message}}", + "db.backend.error.hive_protocol_error": "HiveServer2-Thrift-Protokollantwort konnte nicht gelesen werden (prüfen Sie, ob der Port ein binärer HiveServer2-Thrift-Port ist): {{detail}}", + "db.backend.error.hive_ddl_empty": "Hive hat keine CREATE-Anweisung für „{{name}}“ zurückgegeben", + "db.backend.error.hive_row_edit_unsupported": "Hive-Tabellen unterstützen im Datenraster kein Aktualisieren oder Löschen einzelner Zeilen; führen Sie INSERT oder ALTER TABLE DROP PARTITION im SQL-Editor aus (nur Zeileneinfügungen werden unterstützt)" } diff --git a/shared/i18n/en-US.json b/shared/i18n/en-US.json index 4da7104bf..c08c2cd23 100644 --- a/shared/i18n/en-US.json +++ b/shared/i18n/en-US.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "Other", "plugin.market.anonymous_usage": "Browse and install public plugins without signing in", "plugin.market.ratings_on_website": "Ratings, reviews and account actions are managed on the market website. GoNavi only displays public ratings.", - "plugin.market.open_website": "Open market website" + "plugin.market.open_website": "Open market website", + "datasource.variant.hive.v3": "Hive 3.x: HiveServer2 V10 columnar result protocol with LIMIT / OFFSET paging and external table metadata", + "datasource.variant.hive.v4": "Hive 4.x and later: HiveServer2 V11 columnar result protocol with Iceberg / transactional table and full DDL metadata support", + "connection_modal.step1.hint.hive": "Data warehouse / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "HiveServer2 authentication failed: {{detail}}", + "db.backend.error.hive_query_failed": "Hive query failed (SQLState {{sqlState}}, error code {{code}}): {{message}}", + "db.backend.error.hive_protocol_error": "HiveServer2 Thrift protocol response could not be parsed (verify the port is a HiveServer2 binary Thrift port): {{detail}}", + "db.backend.error.hive_ddl_empty": "Hive returned no CREATE statement for \"{{name}}\"", + "db.backend.error.hive_row_edit_unsupported": "Hive tables do not support single-row update or delete by key in the data grid; run INSERT or ALTER TABLE DROP PARTITION in the SQL editor (only row inserts are supported)" } diff --git a/shared/i18n/ja-JP.json b/shared/i18n/ja-JP.json index 907404d6d..336a36665 100644 --- a/shared/i18n/ja-JP.json +++ b/shared/i18n/ja-JP.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "その他", "plugin.market.anonymous_usage": "公開プラグインはログインせずに閲覧・インストールできます", "plugin.market.ratings_on_website": "評価・レビュー・アカウント操作はマーケットサイトで行います。GoNavi は公開評価の表示のみを行います。", - "plugin.market.open_website": "マーケットサイトを開く" + "plugin.market.open_website": "マーケットサイトを開く", + "datasource.variant.hive.v3": "Hive 3.x：HiveServer2 V10 カラムナ結果プロトコル、LIMIT / OFFSET ページングと外部テーブルメタデータに対応", + "datasource.variant.hive.v4": "Hive 4.x 以降：HiveServer2 V11 カラムナ結果プロトコル、Iceberg / トランザクションテーブルと完全な DDL メタデータに対応", + "connection_modal.step1.hint.hive": "分散データウェアハウス / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "HiveServer2 認証に失敗しました：{{detail}}", + "db.backend.error.hive_query_failed": "Hive クエリが失敗しました（SQLState {{sqlState}}、エラーコード {{code}}）：{{message}}", + "db.backend.error.hive_protocol_error": "HiveServer2 Thrift プロトコル応答を解析できませんでした（ポートが HiveServer2 バイナリ Thrift ポートか確認してください）：{{detail}}", + "db.backend.error.hive_ddl_empty": "Hive から「{{name}}」の CREATE 文が返されませんでした", + "db.backend.error.hive_row_edit_unsupported": "Hive テーブルはデータグリッドでの単一行の更新・削除に対応していません。SQL エディタで INSERT または ALTER TABLE DROP PARTITION を実行してください（行の追加のみ対応）" } diff --git a/shared/i18n/ru-RU.json b/shared/i18n/ru-RU.json index 49a98fd8d..ea40fb017 100644 --- a/shared/i18n/ru-RU.json +++ b/shared/i18n/ru-RU.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "Другие", "plugin.market.anonymous_usage": "Просматривайте и устанавливайте публичные плагины без входа", "plugin.market.ratings_on_website": "Оценки, отзывы и учётная запись управляются на сайте каталога. GoNavi только показывает публичные оценки.", - "plugin.market.open_website": "Открыть сайт каталога" + "plugin.market.open_website": "Открыть сайт каталога", + "datasource.variant.hive.v3": "Hive 3.x: колоночный протокол результатов HiveServer2 V10 с постраничным выводом LIMIT / OFFSET и метаданными внешних таблиц", + "datasource.variant.hive.v4": "Hive 4.x и новее: колоночный протокол результатов HiveServer2 V11 с поддержкой таблиц Iceberg / транзакционных таблиц и полных метаданных DDL", + "connection_modal.step1.hint.hive": "Хранилище данных / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "Ошибка аутентификации HiveServer2: {{detail}}", + "db.backend.error.hive_query_failed": "Ошибка запроса Hive (SQLState {{sqlState}}, код ошибки {{code}}): {{message}}", + "db.backend.error.hive_protocol_error": "Не удалось разобрать ответ протокола HiveServer2 Thrift (убедитесь, что указан бинарный порт Thrift HiveServer2): {{detail}}", + "db.backend.error.hive_ddl_empty": "Hive не вернул оператор CREATE для «{{name}}»", + "db.backend.error.hive_row_edit_unsupported": "Таблицы Hive не поддерживают построчное изменение или удаление в таблице данных; выполните INSERT или ALTER TABLE DROP PARTITION в редакторе SQL (поддерживается только добавление строк)" } diff --git a/shared/i18n/zh-CN.json b/shared/i18n/zh-CN.json index 086610836..3d996f4a9 100644 --- a/shared/i18n/zh-CN.json +++ b/shared/i18n/zh-CN.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "其他", "plugin.market.anonymous_usage": "公开插件无需登录即可浏览和安装", "plugin.market.ratings_on_website": "评分、评价和账号操作统一在市场网站处理，GoNavi 仅展示公开评分。", - "plugin.market.open_website": "打开市场网站" + "plugin.market.open_website": "打开市场网站", + "datasource.variant.hive.v3": "Hive 3.x：使用 HiveServer2 V10 列式结果集协议，支持 LIMIT / OFFSET 分页与外表元数据", + "datasource.variant.hive.v4": "Hive 4.x 及更新版本：使用 HiveServer2 V11 列式结果集协议，支持 Iceberg / 事务表与完整 DDL 元数据", + "connection_modal.step1.hint.hive": "分布式数仓 / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "HiveServer2 认证失败：{{detail}}", + "db.backend.error.hive_query_failed": "Hive 查询执行失败（SQLState {{sqlState}}，错误码 {{code}}）：{{message}}", + "db.backend.error.hive_protocol_error": "HiveServer2 Thrift 协议响应无法解析（请确认端口为 HiveServer2 二进制 Thrift 端口）：{{detail}}", + "db.backend.error.hive_ddl_empty": "Hive 未返回对象「{{name}}」的建表语句", + "db.backend.error.hive_row_edit_unsupported": "Hive 表默认不支持按主键定位单行更新或删除，请在查询编辑器中执行 INSERT / ALTER TABLE DROP PARTITION（仅支持新增行导入）" } diff --git a/shared/i18n/zh-TW.json b/shared/i18n/zh-TW.json index 12e534c30..8de8d7820 100644 --- a/shared/i18n/zh-TW.json +++ b/shared/i18n/zh-TW.json @@ -13457,5 +13457,13 @@ "plugin.market.category.other": "其他", "plugin.market.anonymous_usage": "公開外掛無須登入即可瀏覽與安裝", "plugin.market.ratings_on_website": "評分、評價和帳號操作統一在市集網站處理，GoNavi 僅顯示公開評分。", - "plugin.market.open_website": "開啟市集網站" + "plugin.market.open_website": "開啟市集網站", + "datasource.variant.hive.v3": "Hive 3.x：使用 HiveServer2 V10 列式結果集協定，支援 LIMIT / OFFSET 分頁與外表元資料", + "datasource.variant.hive.v4": "Hive 4.x 及更新版本：使用 HiveServer2 V11 列式結果集協定，支援 Iceberg / 交易表與完整 DDL 元資料", + "connection_modal.step1.hint.hive": "分散式數倉 / HiveServer2 Thrift", + "db.backend.error.hive_auth_failed": "HiveServer2 驗證失敗：{{detail}}", + "db.backend.error.hive_query_failed": "Hive 查詢執行失敗（SQLState {{sqlState}}，錯誤碼 {{code}}）：{{message}}", + "db.backend.error.hive_protocol_error": "HiveServer2 Thrift 協定回應無法解析（請確認連接埠為 HiveServer2 二進位 Thrift 連接埠）：{{detail}}", + "db.backend.error.hive_ddl_empty": "Hive 未傳回物件「{{name}}」的建表語句", + "db.backend.error.hive_row_edit_unsupported": "Hive 表預設不支援按主鍵定位單行更新或刪除，請在查詢編輯器中執行 INSERT / ALTER TABLE DROP PARTITION（僅支援新增行匯入）" }