diff --git a/agents/drivers/hive-go/config.go b/agents/drivers/hive-go/config.go index e18d50c77f..787237556b 100644 --- a/agents/drivers/hive-go/config.go +++ b/agents/drivers/hive-go/config.go @@ -24,6 +24,7 @@ const ( defaultHiveDatabase = "default" defaultHiveHTTPPath = "cliservice" defaultHiveService = "hive" + defaultImpalaService = "impala" defaultZooKeeperNamespace = "hiveserver2" resultSetUniqueColumnNames = "hive.resultset.use.unique.column.names" defaultConnectTimeout = 15 * time.Second @@ -47,6 +48,7 @@ type connectParams struct { ConnectTimeout int `json:"connect_timeout_secs"` AgentJavaOptions []string `json:"agent_java_options"` SessionRole string `json:"sessionRole"` + DatabaseType string `json:"database_type"` } type endpoint struct { @@ -95,6 +97,7 @@ type zooKeeperKerberosConfig struct { } type connectionConfig struct { + DatabaseType string Endpoints []endpoint Database string Username string @@ -138,6 +141,7 @@ type connectionConfig struct { func parseConnectionConfig(params connectParams) (connectionConfig, error) { config := connectionConfig{ + DatabaseType: strings.ToLower(strings.TrimSpace(params.DatabaseType)), Database: strings.TrimSpace(params.Database), Username: params.Username, Password: params.Password, @@ -165,6 +169,10 @@ func parseConnectionConfig(params connectParams) (connectionConfig, error) { CanonicalHostname: true, }, } + if strings.EqualFold(params.DatabaseType, "impala") { + config.Auth = "NOSASL" + config.Kerberos.Service = defaultImpalaService + } if config.Database == "" { config.Database = defaultHiveDatabase } @@ -596,7 +604,7 @@ func applyHiveParameters(config *connectionConfig, values, hiveConfs map[string] parameter(values, "clientprincipal"), parameter(values, "userprincipal"), ) - kerberos.Service = firstNonEmpty(parameter(values, "service"), serviceFromPrincipal(kerberos.ServerPrincipal), defaultHiveService) + kerberos.Service = firstNonEmpty(parameter(values, "service"), serviceFromPrincipal(kerberos.ServerPrincipal), kerberos.Service) kerberos.ServerName = parameter(values, "servername") kerberos.Realm = firstNonEmpty(parameter(values, "realm"), realmFromPrincipal(kerberos.ClientPrincipal)) kerberos.ConfigPath = firstNonEmpty(parameter(values, "krb5conf"), parameter(values, "kerberosconfig")) diff --git a/agents/drivers/hive-go/config_test.go b/agents/drivers/hive-go/config_test.go index c28a448598..5a750378d1 100644 --- a/agents/drivers/hive-go/config_test.go +++ b/agents/drivers/hive-go/config_test.go @@ -178,6 +178,65 @@ func TestURLParamsOverrideConnectionString(t *testing.T) { } } +func TestImpalaDefaultsToNoSASL(t *testing.T) { + config, err := parseConnectionConfig(connectParams{ + DatabaseType: "impala", + Host: "impala.example.com", + Port: 21050, + Database: "analytics", + }) + if err != nil { + t.Fatal(err) + } + if config.Auth != "NOSASL" { + t.Fatalf("unexpected Impala auth mode: %q", config.Auth) + } + if config.Kerberos.Service != "impala" { + t.Fatalf("unexpected Impala Kerberos service: %q", config.Kerberos.Service) + } +} + +func TestImpalaExplicitAuthenticationOverridesDefaults(t *testing.T) { + config, err := parseConnectionConfig(connectParams{ + DatabaseType: "impala", + Host: "impala.example.com", + Port: 21050, + URLParams: "auth=NONE;service=custom", + }) + if err != nil { + t.Fatal(err) + } + if config.Auth != "NONE" || config.Kerberos.Service != "custom" { + t.Fatalf("explicit Impala authentication was not preserved: %#v", config.Kerberos) + } +} + +func TestImpalaLDAPHTTPSSLConfiguration(t *testing.T) { + config, err := parseConnectionConfig(connectParams{ + DatabaseType: "impala", + Host: "impala.example.com", + Port: 21050, + Username: "alice", + Password: "secret", + URLParams: "auth=LDAP;transportMode=http;httpPath=cliservice;ssl=true", + }) + if err != nil { + t.Fatal(err) + } + if config.Auth != "LDAP" || config.TransportMode != "http" || config.HTTPPath != "cliservice" { + t.Fatalf("unexpected Impala LDAP transport config: %#v", config) + } + if config.Username != "alice" || config.Password != "secret" { + t.Fatalf("Impala LDAP credentials were not preserved: %q / %q", config.Username, config.Password) + } + if config.TLSConfig == nil || config.TLSConfig.ServerName != "impala.example.com" { + t.Fatalf("unexpected Impala LDAP TLS config: %#v", config.TLSConfig) + } + if config.Kerberos.Service != defaultImpalaService { + t.Fatalf("unexpected Impala service default: %q", config.Kerberos.Service) + } +} + func TestParseStandardJDBCURLSectionsAndCredentials(t *testing.T) { config, err := parseConnectionConfig(connectParams{ ConnectionString: "jdbc:hive2://hs2.example.com:10001/analytics;user=alice;password=p%40ss?hive.server2.transport.mode=http;hive.server2.thrift.http.path=proxy;hive.exec.dynamic.partition=true#SourceTable=events", diff --git a/agents/drivers/hive-go/connector.go b/agents/drivers/hive-go/connector.go index 865a452fb7..c232d08df8 100644 --- a/agents/drivers/hive-go/connector.go +++ b/agents/drivers/hive-go/connector.go @@ -89,6 +89,7 @@ func newDiscoveryConnector(config connectionConfig) *discoveryConnector { BrowserResponsePort: config.BrowserResponsePort, BrowserResponseTimeout: config.BrowserResponseTimeout, BrowserDisableSSLCheck: config.BrowserDisableSSLCheck, + WaitForNonQueryCompletion: strings.EqualFold(config.DatabaseType, "impala"), }) }, driver: &gohive.Driver{}, diff --git a/agents/drivers/hive-go/metadata.go b/agents/drivers/hive-go/metadata.go index 05c1610fac..8fb1ff97a9 100644 --- a/agents/drivers/hive-go/metadata.go +++ b/agents/drivers/hive-go/metadata.go @@ -197,6 +197,14 @@ func (server *server) connectionInfo() (map[string]any, error) { username = current } } + productName := "Apache Hive" + compatibilityMode := "hive" + driverName := "DBX Hive Go Agent" + if strings.EqualFold(server.params.DatabaseType, "impala") || strings.Contains(strings.ToLower(version), "impalad version") { + productName = "Apache Impala" + compatibilityMode = "impala" + driverName = "DBX Impala Go Agent" + } return map[string]any{ "database": server.config.Database, "schema": server.config.Database, @@ -204,13 +212,13 @@ func (server *server) connectionInfo() (map[string]any, error) { "version": version, "sqlDialect": "HIVE", "identifierQuote": "`", - "compatibilityMode": "hive", + "compatibilityMode": compatibilityMode, "databaseInfo": map[string]string{ - "productName": "Apache Hive", + "productName": productName, "productVersion": version, "unquotedIdentifierCase": "mixed", "quotedIdentifierCase": "mixed", - "driverName": "DBX Hive Go Agent", + "driverName": driverName, "driverVersion": "gohive-v2.1.0", }, }, nil @@ -305,29 +313,119 @@ func (server *server) listTables(schema string, constraints metadataListConstrai sort.Slice(values, func(first, second int) bool { return values[first].Name < values[second].Name }) return applyMetadataWindow(values, constraints.Offset, constraints.Limit), nil } - fallbackStatement := "SHOW TABLES IN " + quoteHiveIdentifier(schema) - if len(requestedTypes) > 0 && !containsString(requestedTypes, "TABLE") { - fallbackStatement = "SHOW VIEWS IN " + quoteHiveIdentifier(schema) + type fallbackQuery struct { + operation string + statement string + objectType string } - result, err := server.executeQuery(queryOptions{ - SQL: fallbackStatement, - MaxRows: metadataQueryLimit, - }) - if err != nil { - return nil, fmt.Errorf("HiveServer2 metadata failed (%v); SHOW TABLES fallback failed: %w", metadataErr, err) + fallbackQueries := make([]fallbackQuery, 0, 2) + if containsString(requestedTypes, "TABLE") { + fallbackQueries = append(fallbackQueries, fallbackQuery{ + operation: "SHOW TABLES", + statement: "SHOW TABLES IN " + quoteHiveIdentifier(schema), + objectType: "TABLE", + }) } - values := make([]tableInfo, 0, len(result.Rows)) - for _, row := range result.Rows { - name := showTablesRowName(result.Columns, row) - if name == "" || !metadataNameMatches(name, constraints.Filter) { - continue + if containsString(requestedTypes, "VIEW") || containsString(requestedTypes, "MATERIALIZED VIEW") { + fallbackQueries = append(fallbackQueries, fallbackQuery{ + operation: "SHOW VIEWS", + statement: "SHOW VIEWS IN " + quoteHiveIdentifier(schema), + objectType: "VIEW", + }) + } + objectsByName := make(map[string]tableInfo) + tableFallbackSucceeded := false + for _, fallback := range fallbackQueries { + result, err := server.executeQuery(queryOptions{SQL: fallback.statement, MaxRows: metadataQueryLimit}) + if err != nil { + // Older Hive and Impala versions can list tables but do not support SHOW VIEWS. + // Keep the usable table result for mixed requests; explicit view requests still fail. + if fallback.objectType == "VIEW" && tableFallbackSucceeded && showViewsUnsupported(err) { + continue + } + return nil, fmt.Errorf("HiveServer2 metadata failed (%v); %s fallback failed: %w", metadataErr, fallback.operation, err) } - values = append(values, tableInfo{Name: name, TableType: "TABLE", Comment: nil}) + if fallback.objectType == "TABLE" { + tableFallbackSucceeded = true + } + for _, row := range result.Rows { + name := showTablesRowName(result.Columns, row) + if name == "" || !metadataNameMatches(name, constraints.Filter) { + continue + } + candidate := tableInfo{Name: name, TableType: fallback.objectType, Comment: nil} + if existing, ok := objectsByName[name]; ok && existing.TableType == "VIEW" && candidate.TableType != "VIEW" { + continue + } + objectsByName[name] = candidate + } + } + values := make([]tableInfo, 0, len(objectsByName)) + for _, value := range objectsByName { + values = append(values, value) } sort.Slice(values, func(first, second int) bool { return values[first].Name < values[second].Name }) return applyMetadataWindow(values, constraints.Offset, constraints.Limit), nil } +func showViewsUnsupported(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + message := strings.ToLower(err.Error()) + operationalMarkers := []string{ + "permission", + "access denied", + "not authorized", + "unauthorized", + "authentication", + "authorization", + "timeout", + "timed out", + "deadline exceeded", + "cancelled", + "canceled", + "transport", + "connection", + "broken pipe", + "network", + } + for _, marker := range operationalMarkers { + if strings.Contains(message, marker) { + return false + } + } + explicitMarkers := []string{ + "unsupported", + "not supported", + "not implemented", + "unknown statement", + "unrecognized statement", + } + for _, marker := range explicitMarkers { + if strings.Contains(message, marker) { + return true + } + } + parseMarkers := []string{ + "parseexception", + "parse error", + "syntax error", + "mismatched input", + "cannot recognize input", + "no viable alternative", + } + for _, marker := range parseMarkers { + if strings.Contains(message, marker) { + return true + } + } + return false +} + func (server *server) listObjects(schema string, constraints metadataListConstraints) ([]objectInfo, error) { if !acceptsHiveTable(constraints.ObjectTypes) { return []objectInfo{}, nil diff --git a/agents/drivers/hive-go/metadata_test.go b/agents/drivers/hive-go/metadata_test.go index 8e988bbccd..c9bba860b9 100644 --- a/agents/drivers/hive-go/metadata_test.go +++ b/agents/drivers/hive-go/metadata_test.go @@ -5,6 +5,7 @@ import ( "database/sql/driver" "errors" "reflect" + "strings" "testing" "github.com/t8y2/dbx/agents/go-common/gohive" @@ -211,7 +212,131 @@ func TestTableCommentAndTypeInfoUseHiveServerMetadata(t *testing.T) { } } -func TestListTablesFallsBackToShowTables(t *testing.T) { +func TestListTablesFallsBackToShowTablesAndViews(t *testing.T) { + behavior := &scriptedBehavior{ + getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { + return gohive.MetadataResult{}, errors.New("metadata unsupported") + }, + query: func(ctx context.Context, query string) (driver.Rows, error) { + switch query { + case "SHOW TABLES IN `analytics`": + return newScriptedRows( + ctx, + []string{"tab_name"}, + []string{"STRING"}, + [][]driver.Value{{"events"}, {"shared_name"}}, + ), nil + case "SHOW VIEWS IN `analytics`": + return newScriptedRows( + ctx, + []string{"view_name"}, + []string{"STRING"}, + [][]driver.Value{{"events_view"}, {"shared_name"}}, + ), nil + default: + t.Fatalf("unexpected fallback query: %q", query) + return nil, errors.New("unexpected fallback query") + } + }, + } + server := newScriptedServer(t, behavior) + defer server.disconnect() + values, err := server.listTables("analytics", metadataListConstraints{}) + if err != nil { + t.Fatal(err) + } + expected := []tableInfo{ + {Name: "events", TableType: "TABLE"}, + {Name: "events_view", TableType: "VIEW"}, + {Name: "shared_name", TableType: "VIEW"}, + } + if !reflect.DeepEqual(values, expected) { + t.Fatalf("unexpected fallback tables: %#v", values) + } +} + +func TestListTablesKeepsShowTablesResultsWhenShowViewsIsUnsupported(t *testing.T) { + behavior := &scriptedBehavior{ + getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { + return gohive.MetadataResult{}, errors.New("metadata unsupported") + }, + query: func(ctx context.Context, query string) (driver.Rows, error) { + switch query { + case "SHOW TABLES IN `analytics`": + return newScriptedRows(ctx, []string{"tab_name"}, []string{"STRING"}, [][]driver.Value{{"events"}}), nil + case "SHOW VIEWS IN `analytics`": + return nil, errors.New("SHOW VIEWS is unsupported") + default: + t.Fatalf("unexpected fallback query: %q", query) + return nil, errors.New("unexpected fallback query") + } + }, + } + server := newScriptedServer(t, behavior) + defer server.disconnect() + + values, err := server.listTables("analytics", metadataListConstraints{}) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(values, []tableInfo{{Name: "events", TableType: "TABLE"}}) { + t.Fatalf("unexpected fallback tables: %#v", values) + } +} + +func TestListTablesReturnsNonCapabilityShowViewsError(t *testing.T) { + behavior := &scriptedBehavior{ + getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { + return gohive.MetadataResult{}, errors.New("metadata unsupported") + }, + query: func(ctx context.Context, query string) (driver.Rows, error) { + switch query { + case "SHOW TABLES IN `analytics`": + return newScriptedRows(ctx, []string{"tab_name"}, []string{"STRING"}, [][]driver.Value{{"events"}}), nil + case "SHOW VIEWS IN `analytics`": + return nil, errors.New("permission denied for SHOW VIEWS") + default: + t.Fatalf("unexpected fallback query: %q", query) + return nil, errors.New("unexpected fallback query") + } + }, + } + server := newScriptedServer(t, behavior) + defer server.disconnect() + + _, err := server.listTables("analytics", metadataListConstraints{}) + if err == nil || !strings.Contains(err.Error(), "SHOW VIEWS fallback failed: permission denied") { + t.Fatalf("unexpected mixed fallback error: %v", err) + } +} + +func TestShowViewsUnsupported(t *testing.T) { + tests := []struct { + name string + err error + unsupported bool + }{ + {name: "explicit unsupported", err: errors.New("SHOW VIEWS is unsupported"), unsupported: true}, + {name: "not supported", err: errors.New("SHOW VIEWS is not supported before Hive 2.2"), unsupported: true}, + {name: "old parser", err: errors.New("ParseException: syntax error at or near VIEWS"), unsupported: true}, + {name: "permission", err: errors.New("permission denied for SHOW VIEWS")}, + {name: "timeout", err: context.DeadlineExceeded}, + {name: "cancel", err: context.Canceled}, + {name: "authentication", err: errors.New("authentication failed")}, + {name: "unsupported authentication", err: errors.New("unsupported authentication mechanism")}, + {name: "transport", err: errors.New("transport is closed")}, + {name: "unsupported transport", err: errors.New("transport does not support SASL")}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if actual := showViewsUnsupported(test.err); actual != test.unsupported { + t.Fatalf("showViewsUnsupported(%v) = %v, want %v", test.err, actual, test.unsupported) + } + }) + } +} + +func TestListTablesFallbackHonorsExplicitTableType(t *testing.T) { behavior := &scriptedBehavior{ getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { return gohive.MetadataResult{}, errors.New("metadata unsupported") @@ -225,7 +350,8 @@ func TestListTablesFallsBackToShowTables(t *testing.T) { } server := newScriptedServer(t, behavior) defer server.disconnect() - values, err := server.listTables("analytics", metadataListConstraints{}) + + values, err := server.listTables("analytics", metadataListConstraints{ObjectTypes: []string{"TABLE"}}) if err != nil { t.Fatal(err) } @@ -234,6 +360,50 @@ func TestListTablesFallsBackToShowTables(t *testing.T) { } } +func TestListViewsFallsBackToShowViews(t *testing.T) { + behavior := &scriptedBehavior{ + getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { + return gohive.MetadataResult{}, errors.New("metadata unsupported") + }, + query: func(ctx context.Context, query string) (driver.Rows, error) { + if query != "SHOW VIEWS IN `analytics`" { + t.Fatalf("unexpected fallback query: %q", query) + } + return newScriptedRows(ctx, []string{"view_name"}, []string{"STRING"}, [][]driver.Value{{"events_view"}}), nil + }, + } + server := newScriptedServer(t, behavior) + defer server.disconnect() + values, err := server.listTables("analytics", metadataListConstraints{ObjectTypes: []string{"VIEW"}}) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(values, []tableInfo{{Name: "events_view", TableType: "VIEW"}}) { + t.Fatalf("unexpected fallback views: %#v", values) + } +} + +func TestListViewsReturnsFallbackErrorWhenShowViewsIsUnsupported(t *testing.T) { + behavior := &scriptedBehavior{ + getTables: func(context.Context, string, string, []string) (gohive.MetadataResult, error) { + return gohive.MetadataResult{}, errors.New("metadata unsupported") + }, + query: func(_ context.Context, query string) (driver.Rows, error) { + if query != "SHOW VIEWS IN `analytics`" { + t.Fatalf("unexpected fallback query: %q", query) + } + return nil, errors.New("SHOW VIEWS is unsupported") + }, + } + server := newScriptedServer(t, behavior) + defer server.disconnect() + + _, err := server.listTables("analytics", metadataListConstraints{ObjectTypes: []string{"VIEW"}}) + if err == nil || !strings.Contains(err.Error(), "SHOW VIEWS fallback failed") { + t.Fatalf("unexpected explicit view fallback error: %v", err) + } +} + func metadataResult(columns []string, rows ...[]driver.Value) gohive.MetadataResult { return gohive.MetadataResult{Columns: columns, Rows: rows} } diff --git a/agents/drivers/hive-go/query_test.go b/agents/drivers/hive-go/query_test.go index 8ae4feeb15..e49cdd5e46 100644 --- a/agents/drivers/hive-go/query_test.go +++ b/agents/drivers/hive-go/query_test.go @@ -338,6 +338,36 @@ func TestQueryResultsPreserveHiveServerLabelsWithoutDotSplitting(t *testing.T) { } } +func TestPagedQueryPreservesDuplicateLeadingValuesAcrossPages(t *testing.T) { + behavior := &scriptedBehavior{} + behavior.query = func(ctx context.Context, _ string) (driver.Rows, error) { + return newScriptedRows( + ctx, + []string{"group_id", "row_id"}, + []string{"BIGINT", "BIGINT"}, + [][]driver.Value{{int64(1), int64(101)}, {int64(1), int64(102)}, {int64(1), int64(103)}}, + ), nil + } + server := newScriptedServer(t, behavior) + + first, err := server.executeQueryPage(queryOptions{SQL: "SELECT group_id, row_id FROM repeated_values", MaxRows: 3}, 2) + if err != nil { + t.Fatal(err) + } + if first.SessionID == nil || !first.HasMore { + t.Fatalf("expected an open cursor after the first page: %#v", first) + } + second, err := server.fetchQueryPage(*first.SessionID, 2) + if err != nil { + t.Fatal(err) + } + allRows := append(append([][]any{}, first.Rows...), second.Rows...) + expected := [][]any{{"1", "101"}, {"1", "102"}, {"1", "103"}} + if !reflect.DeepEqual(allRows, expected) || second.HasMore { + t.Fatalf("duplicate leading values changed across cursor pages: %#v", allRows) + } +} + func TestPagedQueryTruncatesAndPreservesLegacyJDBCValueSemantics(t *testing.T) { largeValue := strings.Repeat("x", 256*1024) createdAt := time.Date(2026, time.August, 11, 10, 11, 12, 345000000, time.UTC) diff --git a/agents/go-common/gohive/connector.go b/agents/go-common/gohive/connector.go index f11e2f8a2d..03d61d9734 100644 --- a/agents/go-common/gohive/connector.go +++ b/agents/go-common/gohive/connector.go @@ -49,6 +49,7 @@ type Config struct { BrowserResponsePort int BrowserResponseTimeout time.Duration BrowserDisableSSLCheck bool + WaitForNonQueryCompletion bool } var _ driver.Connector = (*HiveConnector)(nil) @@ -110,6 +111,7 @@ func (c *HiveConnector) Connect(ctx context.Context) (driver.Conn, error) { connCfg.BrowserResponsePort = c.cfg.BrowserResponsePort connCfg.BrowserResponseTimeout = c.cfg.BrowserResponseTimeout connCfg.BrowserDisableSSLCheck = c.cfg.BrowserDisableSSLCheck + connCfg.WaitForNonQueryCompletion = c.cfg.WaitForNonQueryCompletion // Fallback: build TLS config from cert/key files if TLSConfig not provided directly if connCfg.TLSConfig == nil { diff --git a/agents/go-common/gohive/driver.go b/agents/go-common/gohive/driver.go index 2e47ca3b83..163c7ccf42 100644 --- a/agents/go-common/gohive/driver.go +++ b/agents/go-common/gohive/driver.go @@ -180,6 +180,12 @@ func (c *sqlConnection) QueryContext(ctx context.Context, query string, args []d return nil, fmt.Errorf("HiveServer2 returned no operation handle") } if !cursor.operationHandle.HasResultSet { + if cursor.conn.configuration.WaitForNonQueryCompletion { + if err := cursor.waitForCompletion(ctx); err != nil { + _ = closeCursor(cursor) + return nil, err + } + } result := &NonQueryResult{AffectedRows: cursorAffectedRows(cursor)} if err := closeCursor(cursor); err != nil { return nil, err diff --git a/agents/go-common/gohive/hive.go b/agents/go-common/gohive/hive.go index 2d4d909ae5..7b8ce285f0 100644 --- a/agents/go-common/gohive/hive.go +++ b/agents/go-common/gohive/hive.go @@ -270,6 +270,7 @@ type connectConfiguration struct { BrowserResponsePort int BrowserResponseTimeout time.Duration BrowserDisableSSLCheck bool + WaitForNonQueryCompletion bool // Maximum length of the data in bytes. Used for SASL. MaxSize uint32 MaxMessageSize int32 @@ -1407,6 +1408,54 @@ func (c *cursor) hasMore(ctx context.Context) bool { return c.state != _FINISHED || c.totalRows != c.columnIndex } +func (c *cursor) waitForCompletion(ctx context.Context) error { + if c.operationHandle == nil { + return errors.New("HiveServer2 returned no operation handle") + } + for { + request := hiveserver.NewTGetOperationStatusReq() + request.OperationHandle = c.operationHandle + c.conn.clientMu.Lock() + response, err := c.conn.client.GetOperationStatus(ctx, request) + c.conn.clientMu.Unlock() + if err != nil { + return err + } + if response == nil { + return errors.New("HiveServer2 GetOperationStatus returned no response") + } + if !success(safeStatus(response.GetStatus())) { + return hiveStatusError("checking operation status", response.GetStatus()) + } + switch response.GetOperationState() { + case hiveserver.TOperationState_FINISHED_STATE, hiveserver.TOperationState_CLOSED_STATE: + if response.IsSetNumModifiedRows() { + modified := float64(response.GetNumModifiedRows()) + c.operationHandle.ModifiedRowCount = &modified + } + return nil + case hiveserver.TOperationState_ERROR_STATE: + return &Error{ + Err: errors.New("HiveServer2 operation failed: " + response.GetErrorMessage()), + Message: response.GetErrorMessage(), + ErrorCode: int(response.GetErrorCode()), + SQLState: response.GetSqlState(), + } + case hiveserver.TOperationState_CANCELED_STATE: + return errors.New("HiveServer2 operation was canceled") + case hiveserver.TOperationState_TIMEDOUT_STATE: + return errors.New("HiveServer2 operation timed out") + } + timer := time.NewTimer(time.Duration(c.conn.configuration.PollIntervalInMillis) * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} + func (c *cursor) error() error { return c.Err } diff --git a/apps/desktop/public/icons/database/impala.svg b/apps/desktop/public/icons/database/impala.svg new file mode 100644 index 0000000000..022560e594 --- /dev/null +++ b/apps/desktop/public/icons/database/impala.svg @@ -0,0 +1,3 @@ + + + diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index b35d3cf8e0..d45723cfec 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -1133,6 +1133,7 @@ const driverProfiles: Record< trino: { type: "trino", port: 8080, user: "", label: "Trino", icon: "trino" }, prestosql: { type: "prestosql", port: 8080, user: "", label: "PrestoSQL", icon: "presto" }, hive: { type: "hive", port: 10000, user: "", label: "Apache Hive", icon: "hive" }, + impala: { type: "impala", port: 21050, user: "", label: "Apache Impala", icon: "impala", urlParams: "auth=noSasl" }, spark: { type: "spark", port: 10015, user: "", label: "Apache Spark", icon: "spark" }, db2: { type: "db2", port: 50000, user: "db2inst1", label: "IBM DB2", icon: "db2" }, informix: { type: "informix", port: 9088, user: "informix", label: "Informix", icon: "informix" }, @@ -2386,7 +2387,7 @@ function applyProfile(val: string, preserveConnectionFields = false) { form.value.connection_string = undefined; form.value.url_params = ""; } - resetHiveKerberosFields(profile.type === "hive" ? form.value : undefined); + resetHiveKerberosFields(profile.type === "hive" || profile.type === "impala" ? form.value : undefined); } } @@ -2512,7 +2513,7 @@ watch( resetVictoriaMetricsFields(); } resetElasticsearchProxyFields(config.db_type === "elasticsearch" ? config.external_config : undefined); - resetHiveKerberosFields(config.db_type === "hive" ? config : undefined); + resetHiveKerberosFields(config.db_type === "hive" || config.db_type === "impala" ? config : undefined); resetDamengJvmOptions(config.db_type === "dameng" ? config : undefined); h2ConnectionMode.value = h2ConnectionModeForConfig(config); customColorInput.value = config.color || ""; @@ -2766,6 +2767,7 @@ const iconTypeMap: Record = { trino: "trino", prestosql: "prestosql", hive: "hive", + impala: "impala", spark: "spark", db2: "db2", informix: "informix", @@ -2846,6 +2848,7 @@ const dbOptions: DbOption[] = [ { value: "trino", label: "Trino" }, { value: "prestosql", label: "PrestoSQL" }, { value: "hive", label: "Hive" }, + { value: "impala", label: "Apache Impala" }, { value: "spark", label: "Apache Spark" }, { value: "db2", label: "DB2" }, { value: "informix", label: "Informix" }, @@ -2891,7 +2894,7 @@ const dbCategoryDefinitions: Array<{ { key: "analytics", titleKey: "connection.databaseCategoryAnalytics", - optionValues: ["cloudberry", "clickhouse", "doris", "starrocks", "databend", "selectdb", "databricks", "saphana", "teradata", "vertica", "exasol", "redshift", "snowflake", "trino", "prestosql", "hive", "spark", "bigquery", "kylin", "dremio"], + optionValues: ["cloudberry", "clickhouse", "doris", "starrocks", "databend", "selectdb", "databricks", "saphana", "teradata", "vertica", "exasol", "redshift", "snowflake", "trino", "prestosql", "hive", "impala", "spark", "bigquery", "kylin", "dremio"], }, { key: "domestic", @@ -3728,7 +3731,7 @@ function connectionConfigForSubmit(id: string, generatedName = ""): ConnectionCo config.ssl = !!config.ssl || damengSsl.enabled; config.url_params = applyDamengSslUrlParams(config.url_params, config.ssl, damengSsl.sslFilesPath, damengSsl.sslKeystorePassword, damengSsl.sslProtocol); } - if (config.db_type === "hive") { + if (config.db_type === "hive" || config.db_type === "impala") { if (hiveAuthMode.value === "kerberos" && !hivePrincipal.value.trim()) { throw new Error(t("connection.hiveKerberosPrincipalRequired")); } @@ -6998,7 +7001,7 @@ function openExternalUrl(url: string) {

{{ t("connection.oracleTnsPathHint") }}

-