feat(impala): add full database support

This commit is contained in:
zipg
2026-08-14 20:41:57 +08:00
committed by GitHub
parent 185ff68d78
commit a6d9e6818e
54 changed files with 1092 additions and 148 deletions
+9 -1
View File
@@ -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"))
+59
View File
@@ -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",
+1
View File
@@ -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{},
+116 -18
View File
@@ -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
+172 -2
View File
@@ -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}
}
+30
View File
@@ -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)
+2
View File
@@ -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 {
+6
View File
@@ -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
+49
View File
@@ -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
}
@@ -0,0 +1,3 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 256 613">
<path fill="#2c6791" d="M193.44 15.578c3.46 4.176 6.515 8.262 9.754 12.603 42.171 56.522 54.844 124.223 42.573 193.165-.78 4.384-3.09 4.573-6.364 7.732-20.25 19.539-39.743 39.789-51.848 65.505-5.087 10.809-7.67 22.947-10.146 34.742-.74 3.528 1.775 10.155 4.543 11.362 3.108 1.355 9.634-.494 12.171-3.251 12.905-14.023 25.332-28.525 37.263-43.392 8.141-10.144 15.193-21.162 23.5-31.412 2.246 22.43 1.857 44.475-8.42 64.577-4.325 12.802-9.897 14.317-16.85 22.03-7.912 5.934-17.028 14.959-22.63 22.493-8.753 14.029-8.287 25.364 4.83 35.192.824.504 5.146 2.575 6.546 2.886-2.967 1.986-5.578 2.633-12.377 4.61-.242 10.193-8.736 16.988-11.22 21.752-1.175 4.567-1.613 5.257-1.985 10.349-3.206 43.808-9.676 91.216-25.972 132.447-5.509 13.938-12.149 22.667-25.945 28.445-1.65.692-9.38 4.702-10.374 4.914 3.396-5.394 4.04-13.36-.46-19.473-3.23-4.387-16.119-4.142-20.288-8.155.794-.182 6.325.236 7.112.262 6.572.226 12.79-.107 14.718-8.2 1.546-6.491-1.91-12.974-8.45-17.523-4.444-3.093-7.509-3.971-11.305-7.845-.918-.937-2.304-6.091-1.343-9.49a.64.64 0 0 1 1.046-.316c4.286 3.843 9.878 7.914 13.458 12.194 2.297 2.102 5.195 4.514 7.196 4.489 2.114-.026 3.331-2.596 5.803-4.341 5.691-5.065 7.71-7.532 11.293-14.454.621-3.807-.73-10.268-1.92-14.958-4.982-19.62-6.557-40.03-1.437-59.714 3.67-14.108 8.92-28.515 16.755-40.666 12.532-19.433 13.111-37.213-.019-55.823-22.442-31.81-19.86-63.956-1.676-96.556 12.958-23.233 29.63-43.454 48.09-62.451 3.82-3.932 6.313-4.333 7.368-9.716 13.515-62.3 5.201-125.481-24.557-180.882-1.318-2.39-12.44-17.833-12.608-18.706.35-.235 13.824 15.81 14.174 15.575M71.74 352.47c-.72-9.112-.81-17.71-2.168-26.102-5.42-33.498-23.178-60.454-45.848-84.682-1.8-1.925-3.462-4.106-5.597-5.571-11.324-7.771-14.777-19.198-16.244-32.1C-6.29 132.114 11.98 68.34 59.748 13.483 60.066 13.117 69.524.777 71.066.168 57.671 23.513 41.73 46.315 34.17 85.336c-7.319 37.779-7.397 75.247.506 112.762.807 3.83 3.615 7.639 6.414 10.585 19.841 20.887 38.522 42.783 49.906 69.46 5.46 12.79 9.142 26.845 10.724 40.659 2.146 18.74-9.48 30.81-29.978 33.668m-20.17 58.968c18.106-12.047 31.533-8.303 39.11 8.113 8.06 17.462 13.185 42.644 14.068 61.888-4.56-11.154-10.893-28.827-15.34-40.026-2.09-5.265-4.85-9.09-11.323-8.972-6.052.11-10.041-3.434-11.351-9.1-.602-2.605-1.323-6.505-2.903-8.268-2.633-2.938-6.985-2.996-12.261-3.635m-6.315-57.26C21.625 338.361 4.53 318.096 1.293 287.892c13.006 5.58 36.592 40.667 43.963 66.286"/>
</svg>

After

Width:  |  Height:  |  Size: 2.4 KiB

@@ -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<string, string> = {
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) {
<p class="col-span-3 text-xs text-muted-foreground">{{ t("connection.oracleTnsPathHint") }}</p>
</div>
<template v-if="form.db_type === 'hive'">
<template v-if="form.db_type === 'hive' || form.db_type === 'impala'">
<div class="grid grid-cols-4 items-center gap-4">
<Label :class="connectionLabelClass">{{ t("connection.hiveAuthMode") }}</Label>
<div class="col-span-3 grid h-8 grid-cols-2 overflow-hidden rounded-md border border-input bg-muted/30 p-0.5">
@@ -7020,7 +7023,7 @@ function openExternalUrl(url: string) {
<template v-if="hiveAuthMode === 'kerberos'">
<div class="grid grid-cols-4 items-center gap-4">
<Label :class="connectionLabelSmallClass">{{ t("connection.hivePrincipal") }}</Label>
<Input v-model="hivePrincipal" class="col-span-3" placeholder="hive/_HOST@EXAMPLE.COM" />
<Input v-model="hivePrincipal" class="col-span-3" :placeholder="form.db_type === 'impala' ? 'impala/_HOST@EXAMPLE.COM' : 'hive/_HOST@EXAMPLE.COM'" />
</div>
<div class="grid grid-cols-4 items-center gap-4">
<Label :class="connectionLabelSmallClass">krb5.conf</Label>
@@ -31,7 +31,7 @@ watch(
);
function submit() {
if (resolving.value || !password.value) return;
if (resolving.value) return;
resolving.value = true;
promptStore.submit(password.value, rememberPassword.value);
}
@@ -67,7 +67,7 @@ function cancel() {
<Button variant="outline" :disabled="resolving" @click="cancel">
{{ t("connection.promptPasswordCancel") }}
</Button>
<Button :disabled="resolving || !password" @click="submit">
<Button :disabled="resolving" @click="submit">
{{ t("connection.promptPasswordSubmit") }}
</Button>
</DialogFooter>
+13
View File
@@ -31,6 +31,19 @@ describe("DatabaseIcon", () => {
app.unmount();
});
it("uses the Apache Impala asset instead of the Hive asset", async () => {
const container = document.createElement("div");
document.body.appendChild(container);
const app = createApp(DatabaseIcon, { dbType: "impala" });
app.mount(container);
await nextTick();
const icon = container.querySelector("img");
expect(icon?.getAttribute("src")).toBe("/icons/database/impala.svg");
expect(icon?.classList.contains("database-logo-impala")).toBe(true);
app.unmount();
});
it("uses the Meilisearch asset", async () => {
const container = document.createElement("div");
document.body.appendChild(container);
@@ -55,6 +55,7 @@ const assetIcons: Record<string, string> = {
presto: "presto",
prestosql: "presto",
hive: "hive",
impala: "impala",
hbase: "hbase",
phoenix: "phoenix",
spark: "spark-logo.png",
@@ -122,7 +123,7 @@ const assetSrc = computed(() => {
</script>
<template>
<img v-if="assetName" :src="assetSrc" alt="" class="database-logo object-contain" :class="{ 'database-logo-light': useLightIconInDarkMode }" aria-hidden="true" />
<img v-if="assetName" :src="assetSrc" alt="" class="database-logo object-contain" :class="{ 'database-logo-light': useLightIconInDarkMode, 'database-logo-impala': normalizedType === 'impala' }" aria-hidden="true" />
<Database v-else class="text-blue-400" />
</template>
@@ -135,4 +136,8 @@ const assetSrc = computed(() => {
.database-logo-light {
filter: brightness(0) invert(82%);
}
.database-logo-impala {
transform: scale(1.55);
}
</style>
+2 -2
View File
@@ -448,7 +448,7 @@ export default {
disconnectAndForgetPassword: "Disconnect and Forget Password",
passwordForgotten: "Password forgotten. Reconnect will ask for it again.",
promptPasswordTitle: "Enter password",
promptPasswordMessage: "Enter the password for connection “{connection}”.",
promptPasswordMessage: "Enter the password for connection “{connection}”. For an empty password, connect directly.",
promptPasswordPlaceholder: "Password",
promptPasswordSubmit: "Connect",
promptPasswordCancel: "Cancel",
@@ -510,7 +510,7 @@ export default {
hiveAuthMode: "Auth",
hiveAuthNone: "None",
hivePrincipal: "Principal",
hiveKerberosPrincipalRequired: "Hive Kerberos principal is required",
hiveKerberosPrincipalRequired: "Kerberos principal is required",
hiveKrb5ConfBrowse: "Choose krb5.conf",
hiveJaasConfigBrowse: "Choose JAAS config",
hiveTicketCache: "Ticket Cache",
+1 -1
View File
@@ -449,7 +449,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "Desconectar y olvidar la contraseña",
passwordForgotten: "Contraseña olvidada. La próxima conexión la solicitará de nuevo.",
promptPasswordTitle: "Introducir contraseña",
promptPasswordMessage: "Introduzca la contraseña de la conexión “{connection}”.",
promptPasswordMessage: "Introduzca la contraseña de la conexión “{connection}”. Si está vacía, conéctese directamente.",
promptPasswordPlaceholder: "Contraseña",
promptPasswordSubmit: "Conectar",
promptPasswordCancel: "Cancelar",
+1 -1
View File
@@ -448,7 +448,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "Disconnetti e dimentica la password",
passwordForgotten: "Password dimenticata. La prossima connessione la richiederà di nuovo.",
promptPasswordTitle: "Inserisci password",
promptPasswordMessage: "Inserisci la password per la connessione “{connection}”.",
promptPasswordMessage: "Inserisci la password per la connessione “{connection}”. Se è vuota, connettiti direttamente.",
promptPasswordPlaceholder: "Password",
promptPasswordSubmit: "Connetti",
promptPasswordCancel: "Annulla",
+2 -2
View File
@@ -448,7 +448,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "切断してこのパスワードを忘れる",
passwordForgotten: "パスワードを忘れました。次回の接続で再入力が必要です。",
promptPasswordTitle: "パスワードを入力",
promptPasswordMessage: "接続「{connection}」のパスワードを入力してください。",
promptPasswordMessage: "接続「{connection}」のパスワードを入力してください。空のパスワードの場合は、そのまま接続できます。",
promptPasswordPlaceholder: "パスワード",
promptPasswordSubmit: "接続",
promptPasswordCancel: "キャンセル",
@@ -1104,7 +1104,7 @@ export default withEnglishFallback({
hiveAuthMode: "認証",
hiveAuthNone: "なし",
hivePrincipal: "プリンシパル",
hiveKerberosPrincipalRequired: "Hive Kerberos プリンシパルは必須です",
hiveKerberosPrincipalRequired: "Kerberos プリンシパルは必須です",
hiveKrb5ConfBrowse: "krb5.conf を選択",
hiveJaasConfigBrowse: "JAAS 設定を選択",
hiveTicketCache: "チケットキャッシュ",
+2 -2
View File
@@ -442,7 +442,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "연결 끊고 비밀번호 잊기",
passwordForgotten: "비밀번호를 잊었습니다. 다음 연결 시 다시 입력해야 합니다.",
promptPasswordTitle: "비밀번호 입력",
promptPasswordMessage: "연결 “{connection}”의 비밀번호를 입력하세요.",
promptPasswordMessage: "연결 “{connection}”의 비밀번호를 입력하세요. 빈 비밀번호라면 바로 연결할 수 있습니다.",
promptPasswordPlaceholder: "비밀번호",
promptPasswordSubmit: "연결",
promptPasswordCancel: "취소",
@@ -504,7 +504,7 @@ export default withEnglishFallback({
hiveAuthMode: "인증",
hiveAuthNone: "없음",
hivePrincipal: "프린시펄",
hiveKerberosPrincipalRequired: "Hive Kerberos 프린시펄은 필수입니다",
hiveKerberosPrincipalRequired: "Kerberos 프린시펄은 필수입니다",
hiveKrb5ConfBrowse: "krb5.conf 선택",
hiveJaasConfigBrowse: "JAAS 구성 선택",
hiveTicketCache: "티켓 캐시",
+1 -1
View File
@@ -449,7 +449,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "Desconectar e Esquecer a Senha",
passwordForgotten: "Senha esquecida. A próxima conexão pedirá novamente.",
promptPasswordTitle: "Digite a senha",
promptPasswordMessage: "Digite a senha da conexão “{connection}”.",
promptPasswordMessage: "Digite a senha da conexão “{connection}”. Se estiver vazia, conecte-se diretamente.",
promptPasswordPlaceholder: "Senha",
promptPasswordSubmit: "Conectar",
promptPasswordCancel: "Cancelar",
+2 -2
View File
@@ -373,7 +373,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "断开并忘记本次密码",
passwordForgotten: "已忘记密码,下次连接需重新输入。",
promptPasswordTitle: "输入密码",
promptPasswordMessage: "请输入连接“{connection}”的密码。",
promptPasswordMessage: "请输入连接“{connection}”的密码;若密码为空,可直接连接",
promptPasswordPlaceholder: "密码",
promptPasswordSubmit: "连接",
promptPasswordCancel: "取消",
@@ -435,7 +435,7 @@ export default withEnglishFallback({
hiveAuthMode: "认证",
hiveAuthNone: "无",
hivePrincipal: "Principal",
hiveKerberosPrincipalRequired: "Hive Kerberos principal 不能为空",
hiveKerberosPrincipalRequired: "Kerberos principal 不能为空",
hiveKrb5ConfBrowse: "选择 krb5.conf",
hiveJaasConfigBrowse: "选择 JAAS 配置",
hiveTicketCache: "票据缓存",
+1 -1
View File
@@ -449,7 +449,7 @@ export default withEnglishFallback({
disconnectAndForgetPassword: "中斷連線並忘記本次密碼",
passwordForgotten: "已忘記密碼,下次連線需重新輸入。",
promptPasswordTitle: "輸入密碼",
promptPasswordMessage: "請輸入連線「{connection}」的密碼。",
promptPasswordMessage: "請輸入連線「{connection}」的密碼;若密碼為空,可直接連線",
promptPasswordPlaceholder: "密碼",
promptPasswordSubmit: "連線",
promptPasswordCancel: "取消",
@@ -18,6 +18,10 @@ describe("connectionAttemptTimeout", () => {
expect(connectionAttemptTimeoutMs({ db_type: "prestosql", connect_timeout_secs: 5, transport_layers: [] })).toBe(AGENT_DRIVER_MIN_CONNECT_TIMEOUT_SECS * 1000 + CONNECTION_ATTEMPT_TIMEOUT_BUFFER_MS);
});
it("uses the startup floor for Impala agent connections", () => {
expect(connectionAttemptTimeoutMs({ db_type: "impala", connect_timeout_secs: 5, transport_layers: [] })).toBe(AGENT_DRIVER_MIN_CONNECT_TIMEOUT_SECS * 1000 + CONNECTION_ATTEMPT_TIMEOUT_BUFFER_MS);
});
it("uses the startup floor for generic JDBC plugin connections", () => {
expect(connectionAttemptTimeoutMs({ db_type: "jdbc", connect_timeout_secs: 5, transport_layers: [] })).toBe(AGENT_DRIVER_MIN_CONNECT_TIMEOUT_SECS * 1000 + CONNECTION_ATTEMPT_TIMEOUT_BUFFER_MS);
});
@@ -0,0 +1,36 @@
import { describe, expect, it } from "vitest";
import { connectionNeedsPasswordPrompt, connectionUsesPasswordlessAuthentication } from "@/lib/connection/connectionPassword";
import type { ConnectionConfig } from "@/types/database";
function connection(overrides: Partial<ConnectionConfig> = {}): ConnectionConfig {
return {
id: "connection-1",
name: "Connection",
db_type: "impala",
host: "127.0.0.1",
port: 21050,
username: "",
password: "",
save_password: false,
...overrides,
} as ConnectionConfig;
}
describe("connection password prompting", () => {
it("treats Impala's default and explicit NOSASL modes as passwordless", () => {
expect(connectionUsesPasswordlessAuthentication(connection())).toBe(true);
expect(connectionUsesPasswordlessAuthentication(connection({ url_params: "transportMode=binary;auth=noSasl" }))).toBe(true);
expect(connectionNeedsPasswordPrompt(connection({ url_params: "AUTH=NOSASL&fetchSize=1000" }))).toBe(false);
});
it("still prompts for Impala authentication modes that can require credentials", () => {
expect(connectionNeedsPasswordPrompt(connection({ url_params: "auth=LDAP" }))).toBe(true);
expect(connectionNeedsPasswordPrompt(connection({ url_params: "principal=impala/_HOST@EXAMPLE.COM" }))).toBe(true);
expect(connectionNeedsPasswordPrompt(connection({ url_params: "auth=noSasl;principal=impala/_HOST@EXAMPLE.COM" }))).toBe(true);
});
it("does not change password prompting for other database types", () => {
expect(connectionNeedsPasswordPrompt(connection({ db_type: "mysql", port: 3306, url_params: "auth=noSasl" }))).toBe(true);
expect(connectionNeedsPasswordPrompt(connection({ db_type: "postgres", port: 5432, save_password: true }))).toBe(false);
});
});
@@ -49,6 +49,12 @@ describe("tableEditing", () => {
expect(isTableDataEditable("oracle", [DBX_ROWID_COLUMN], "VIEW")).toBe(false);
});
it("keeps Impala table data readonly", () => {
expect(isTableDataEditable("impala", ["id"], "TABLE")).toBe(false);
expect(canEditExistingTableRows("impala", undefined, ["id"])).toBe(false);
expect(supportsDataGridTransaction("impala")).toBe(false);
});
it("does not include Oracle ROWID for view data tabs", () => {
expect(usesSyntheticRowIdKey("oracle", [DBX_ROWID_COLUMN], "VIEW")).toBe(false);
expect(usesSyntheticRowIdKey("oracle", [DBX_ROWID_COLUMN], "MATERIALIZED_VIEW")).toBe(false);
@@ -9,6 +9,13 @@ describe("DuckDB driver installation", () => {
});
});
describe("Impala driver installation", () => {
it("reuses the downloadable Hive driver", () => {
expect(agentDriverInstallKey("impala")).toBe("hive");
expect(showAgentDriverInstallHint("impala", [{ db_type: "hive", installed: true }])).toBe(false);
});
});
describe("driverStoreFocusForInstallError", () => {
it("focuses the missing agent driver for driver-not-installed errors", () => {
expect(driverStoreFocusForInstallError("zookeeper driver is not installed. Please install it from the Driver Manager.", "zookeeper", undefined)).toEqual({
@@ -24,6 +24,7 @@ export function hasInstalledAgentVersion(drivers: readonly AgentDriverInstallSta
}
export function agentDriverInstallKey(dbType: DatabaseType | undefined, driverProfile?: string): string | undefined {
if (dbType === "impala") return "hive";
if (dbType === "oracle") return "oracle";
if (dbType === "mongodb") return "mongodb";
if (dbType === "dameng") return "dameng";
@@ -30,6 +30,7 @@ const DRIVER_STARTUP_FLOOR_TYPES = new Set<DatabaseType>([
"prestosql",
"jdbc",
"hive",
"impala",
"spark",
"db2",
"informix",
@@ -0,0 +1,30 @@
import type { ConnectionConfig } from "@/types/database";
type PasswordAuthenticationConfig = Pick<ConnectionConfig, "db_type" | "driver_profile" | "url_params">;
function hiveServerUrlParams(urlParams?: string): Map<string, string> {
const params = new Map<string, string>();
for (const part of (urlParams || "").replace(/^[?&;]+|[?&;]+$/g, "").split(/[;&]/)) {
const trimmed = part.trim();
if (!trimmed) continue;
const equals = trimmed.indexOf("=");
const key = (equals < 0 ? trimmed : trimmed.slice(0, equals)).trim().toLowerCase();
const value = (equals < 0 ? "" : trimmed.slice(equals + 1)).trim();
if (key) params.set(key, value);
}
return params;
}
export function connectionUsesPasswordlessAuthentication(config: PasswordAuthenticationConfig): boolean {
const profile = config.driver_profile || config.db_type;
if (config.db_type !== "impala" && profile !== "impala") return false;
const params = hiveServerUrlParams(config.url_params);
if (params.get("principal")) return false;
const auth = (params.get("auth") || "nosasl").toLowerCase();
return auth === "nosasl";
}
export function connectionNeedsPasswordPrompt(config: ConnectionConfig): boolean {
return config.save_password === false && !config.password && !connectionUsesPasswordlessAuthentication(config);
}
@@ -4,7 +4,7 @@ import { manifestDatabaseTypes } from "@/lib/database/databaseDriverManifest";
import { supportsTransfer } from "@/lib/database/databaseFeatureSupport";
import type { DatabaseType } from "@/types/database";
const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "questdb", "hive", "spark"];
const TABLE_ONLY_TRANSFER_DATABASES: DatabaseType[] = ["sqlite", "rqlite", "cloudflare-d1", "duckdb", "clickhouse", "mongodb", "questdb", "hive", "impala", "spark"];
describe("transferObjectKinds", () => {
it("groups databases into transfer families", () => {
@@ -27,6 +27,7 @@ export const SCHEMA_AWARE_TYPES = new Set<DatabaseType>([
"trino",
"prestosql",
"hive",
"impala",
"spark",
"databend",
"db2",
@@ -179,7 +179,7 @@ export function supportsObjectBrowserTreeNode(dbType: DatabaseType | undefined,
}
export function supportsTableTruncate(dbType?: DatabaseType): boolean {
return !!dbType && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "cloudflare-d1" && dbType !== "duckdb" && dbType !== "influxdb" && dbType !== "victoriametrics" && dbType !== "manticoresearch";
return !!dbType && dbType !== "impala" && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "cloudflare-d1" && dbType !== "duckdb" && dbType !== "influxdb" && dbType !== "victoriametrics" && dbType !== "manticoresearch";
}
export function usesPostgresLikeStructureCopy(dbType?: DatabaseType): boolean {
@@ -66,6 +66,7 @@ export const DATABASE_NAMESPACE_CREATION_MATRIX = {
trino: { database: "schema" },
prestosql: { database: "schema" },
hive: { deferred: "Hive database creation needs agent metadata validation first" },
impala: { deferred: "Impala database creation needs dedicated metadata validation first" },
spark: { deferred: "Spark database creation needs agent metadata validation first" },
db2: { database: "schema" },
informix: { database: "schema" },
@@ -69,6 +69,7 @@ const DATABASE_TYPE_OBJECTS = new Map<DatabaseType, SidebarObjectKind[]>([
["doris", TABLE_VIEW_OBJECTS],
["starrocks", TABLE_VIEW_MV_OBJECTS],
["hive", TABLE_VIEW_OBJECTS],
["impala", TABLE_VIEW_OBJECTS],
["spark", TABLE_VIEW_OBJECTS],
["trino", TABLE_VIEW_OBJECTS],
["prestosql", TABLE_VIEW_OBJECTS],
@@ -66,6 +66,7 @@ export const DATABASE_PROPERTY_EDITING_MATRIX = {
trino: { deferred: "catalog properties are connector-managed" },
prestosql: { deferred: "catalog properties are connector-managed" },
hive: { deferred: "database properties need agent metadata validation first" },
impala: { deferred: "database properties need Impala-specific validation first" },
spark: { deferred: "database properties need agent metadata validation first" },
db2: { deferred: "schema properties need product-specific handling" },
informix: { deferred: "schema properties need product-specific handling" },
@@ -90,6 +90,17 @@ const NAVICAT_STYLE_TABLE_DATA_TYPES = new Set<DatabaseType>([
]);
const DATABASE_CAPABILITY_OVERRIDES: Partial<Record<DatabaseType, Partial<DatabaseCapability>>> = {
impala: {
tableData: {
insert: false,
updateRequiresPrimaryKey: false,
deleteRequiresPrimaryKey: false,
requiresTransactionalTableForExistingRows: false,
existingRowsReadonly: true,
transaction: false,
readonly: true,
},
},
hive: {
tableData: {
insert: true,
+1 -1
View File
@@ -11,7 +11,7 @@ const GAUSSDB_IDENTIFIER_QUOTE_STYLE_KEY = "gaussdbIdentifierQuoteStyle";
export const GAUSSDB_M_JDBC_DRIVER_PROFILE = "gaussdb-m";
export const GAUSSDB_M_JDBC_DRIVER_CLASS = "com.huawei.gaussdb.jdbc.Driver";
const DATABASE_AS_EXECUTION_SCHEMA_TYPES = new Set<DatabaseType>(["hive", "spark"]);
const DATABASE_AS_EXECUTION_SCHEMA_TYPES = new Set<DatabaseType>(["hive", "impala", "spark"]);
const CONNECTION_ROOT_SCHEMA_TYPES = new Set<DatabaseType>(["oracle", "dameng", "oceanbase-oracle"]);
const JDBC_DIALECT_MATCHERS: Array<{ type: DatabaseType; patterns: RegExp[] }> = [
+1 -1
View File
@@ -33,7 +33,7 @@ export function quoteTableIdentifier(databaseType: DatabaseType | undefined, nam
// (DatabaseMetaData.getIdentifierQuoteString()) — pass through unquoted.
if (databaseType === "jdbc") return name;
if (databaseType === "bigquery") return `\`${name.replace(/`/g, "\\`")}\``;
if (databaseType === "mysql" || databaseType === "clickhouse" || databaseType === "hive" || databaseType === "spark" || databaseType === "databend" || databaseType === "tdengine" || databaseType === "access" || databaseType === "doris" || databaseType === "starrocks")
if (databaseType === "mysql" || databaseType === "clickhouse" || databaseType === "hive" || databaseType === "impala" || databaseType === "spark" || databaseType === "databend" || databaseType === "tdengine" || databaseType === "access" || databaseType === "doris" || databaseType === "starrocks")
return `\`${name.replace(/`/g, "``")}\``;
if (databaseType === "informix" && /^[A-Za-z_][A-Za-z0-9_$]*$/.test(name)) return name;
if (databaseType === "neo4j") return quoteCypherIdentifier(name);
@@ -130,6 +130,62 @@ describe("connectionStore save_password opt-out", () => {
expect(store.getConfig("pg-1")?.password).toBe("");
});
it("connects with an explicitly submitted empty password", async () => {
installApiMocks();
installPasswordPromptMock();
requestPassword.mockResolvedValue({ password: "", rememberPassword: false });
const { useConnectionStore } = await import("@/stores/connectionStore");
const store = useConnectionStore();
const connection = postgresConnection({ id: "pg-1", save_password: false, password: "" });
store.connections = [connection];
await store.connect(connection);
const { connectDb } = await import("@/lib/backend/api");
expect(connectDb).toHaveBeenCalledWith(expect.objectContaining({ id: "pg-1", password: "" }), expect.any(Number));
expect(store.getConfig("pg-1")).toEqual(expect.objectContaining({ password: "", save_password: false }));
});
it("connects to NOSASL Impala without prompting for an empty password", async () => {
installApiMocks();
installPasswordPromptMock();
const { useConnectionStore } = await import("@/stores/connectionStore");
const store = useConnectionStore();
const connection = postgresConnection({
id: "impala-1",
name: "Impala",
db_type: "impala",
port: 21050,
username: "",
password: "",
save_password: false,
url_params: "auth=noSasl",
});
store.connections = [connection];
await store.connect(connection);
expect(requestPassword).not.toHaveBeenCalled();
const { connectDb } = await import("@/lib/backend/api");
expect(connectDb).toHaveBeenCalledWith(expect.objectContaining({ id: "impala-1", password: "", save_password: false }), expect.any(Number));
});
it("remembers an explicitly submitted empty password", async () => {
installApiMocks();
installPasswordPromptMock();
requestPassword.mockResolvedValue({ password: "", rememberPassword: true });
const { useConnectionStore } = await import("@/stores/connectionStore");
const store = useConnectionStore();
const connection = postgresConnection({ id: "pg-1", save_password: false, password: "" });
store.connections = [connection];
await store.connect(connection);
const { saveConnections } = await import("@/lib/backend/api");
expect(saveConnections).toHaveBeenLastCalledWith(expect.arrayContaining([expect.objectContaining({ id: "pg-1", password: "", save_password: true })]));
expect(store.getConfig("pg-1")).toEqual(expect.objectContaining({ password: "", save_password: true }));
});
it("connect cancels when the prompt is dismissed", async () => {
installApiMocks();
installPasswordPromptMock();
+12 -18
View File
@@ -116,6 +116,7 @@ import { REDIS_SCAN_PAGE_SIZE_DEFAULT } from "@/lib/redis/redisKeyPattern";
import { normalizeRedisDatabaseAliases, redisDatabaseAlias, redisDatabaseLabel } from "@/lib/redis/redisDatabaseAlias";
import { appendAgentDriverUpdateHint, hasAgentDriverUpdate, hasInstalledAgentVersion, type AgentDriverInstallState } from "@/lib/connection/agentDriverInstallHint";
import { appendConnectionErrorHints, isMysqlMissingPasswordFailure } from "@/lib/connection/connectionErrorHints";
import { connectionNeedsPasswordPrompt } from "@/lib/connection/connectionPassword";
import { appendVisibleDatabaseSelection } from "@/lib/connection/connectionVisibleDatabases";
import { buildXuguTypeMemberNodes, isXuguTypeMemberContainer } from "@/lib/sidebar/xuguTypeMembers";
import { isXuguPublicSynonymScope, xuguSchemaDisplayName, XUGU_PUBLIC_SYNONYM_SCOPE } from "@/lib/sidebar/xuguPublicSynonyms";
@@ -1055,6 +1056,7 @@ export const useConnectionStore = defineStore("connection", () => {
trino: "Trino",
prestosql: "PrestoSQL",
hive: "Hive",
impala: "Apache Impala",
spark: "Apache Spark",
db2: "DB2",
informix: "Informix",
@@ -3084,12 +3086,12 @@ export const useConnectionStore = defineStore("connection", () => {
connectionId: config.id,
connectionName: config.name,
});
if (!result?.password) throw new Error(CONNECTION_PASSWORD_REQUIRED_MESSAGE);
if (!result) throw new Error(CONNECTION_PASSWORD_REQUIRED_MESSAGE);
return { config: { ...config, password: result.password }, rememberPassword: result.rememberPassword };
}
async function persistRememberedConnectionPassword(config: ConnectionConfig, rememberPassword: boolean, expectedConfigFingerprint: string): Promise<void> {
if (!rememberPassword || !config.password) return;
if (!rememberPassword) return;
const index = connections.value.findIndex((connection) => connection.id === config.id);
if (index < 0) return;
if (connectionConfigFingerprint(connections.value[index]) !== expectedConfigFingerprint) return;
@@ -3125,14 +3127,10 @@ export const useConnectionStore = defineStore("connection", () => {
const localAttempt = beginLocalConnectionAttempt(config.id);
try {
let rememberPassword = false;
if (config.save_password === false && !config.password) {
// 本次运行期已在会话凭据仓库暂存密码(save_password=false):免弹窗,
// 直接以空密码 connectDb,后端会从会话凭据仓库补主密码。
if (!(await hasSessionCredential(config.id))) {
const prompted = await ensureConnectionPassword(config);
config = prompted.config;
rememberPassword = prompted.rememberPassword;
}
if (connectionNeedsPasswordPrompt(config) && !(await hasSessionCredential(config.id))) {
const prompted = await ensureConnectionPassword(config);
config = prompted.config;
rememberPassword = prompted.rememberPassword;
}
await beforeConnectHandler?.(config);
if (config.db_type === "sqlserver") {
@@ -3340,14 +3338,10 @@ export const useConnectionStore = defineStore("connection", () => {
// Fast-path the common case (password saved or no password needed) so the
// in-flight dedup above keeps its exact microtask cadence; only await the
// interactive prompt when the connection actually needs a typed password.
if (config.save_password === false && !config.password) {
// 本次运行期已在会话凭据仓库暂存密码(save_password=false):免弹窗,
// 直接以空密码 connectDb,后端会从会话凭据仓库补主密码。
if (!(await hasSessionCredential(connectionId))) {
const prompted = await ensureConnectionPassword(config);
config = prompted.config;
rememberPassword = prompted.rememberPassword;
}
if (connectionNeedsPasswordPrompt(config) && !(await hasSessionCredential(connectionId))) {
const prompted = await ensureConnectionPassword(config);
config = prompted.config;
rememberPassword = prompted.rememberPassword;
}
await beforeConnectHandler?.(config);
if (config.db_type === "sqlserver") {
+1
View File
@@ -53,6 +53,7 @@ export type DatabaseType =
| "trino"
| "prestosql"
| "hive"
| "impala"
| "spark"
| "db2"
| "informix"
@@ -1460,6 +1460,36 @@
"driverManagement": true
}
},
{
"dbType": "impala",
"label": "Apache Impala",
"runtimeMode": "agent",
"mcpMode": "bridge",
"agentKey": "hive",
"singleConnectionPool": false,
"metadataConnectionScoped": false,
"skipTcpProbe": true,
"defaultPort": 21050,
"supportLevel": "operate",
"capabilities": {
"queryExecution": true,
"metadataBrowse": true,
"objectBrowser": true,
"objectSource": true,
"schemaSearch": true,
"diagram": false,
"tableDataEdit": false,
"tableStructureEdit": false,
"tableImport": false,
"dataTransfer": true,
"sqlFileExecution": true,
"databaseCreate": false,
"fieldLineage": false,
"sqlExplain": true,
"userAdmin": false,
"driverManagement": true
}
},
{
"dbType": "db2",
"label": "IBM DB2",
+13
View File
@@ -196,6 +196,13 @@ const AGENT_CATALOG: &[AgentCatalogEntry] = &[
store_visible: true,
profiles: &[],
},
AgentCatalogEntry {
db_type: DatabaseType::Impala,
key: "hive",
label: "Apache Impala",
store_visible: false,
profiles: &[],
},
AgentCatalogEntry {
db_type: DatabaseType::Spark,
key: "spark",
@@ -397,4 +404,10 @@ mod tests {
assert_eq!(label_for_key("duckdb"), Some("DuckDB"));
assert!(!is_agent_type(&DatabaseType::DuckDb));
}
#[test]
fn impala_reuses_hive_agent_without_duplicate_store_entry() {
assert_eq!(agent_key(&DatabaseType::Impala, None), Some("hive"));
assert_eq!(driver_store_entries().filter(|(key, _)| *key == "hive").count(), 1);
}
}
+2
View File
@@ -110,6 +110,7 @@ pub fn agent_connect_params_with_role(
"jdbc_driver_class": agent_jdbc_driver_class(config),
"jdbc_driver_paths": &config.jdbc_driver_paths,
"sessionRole": session_role.as_str(),
"database_type": config.db_type,
});
if config.db_type == DatabaseType::ZooKeeper {
params["connection_timeout_ms"] = serde_json::json!(
@@ -700,6 +701,7 @@ mod tests {
fn agent_connect_params_default_to_workload_session_role() {
let params = agent_connect_params(&config(DatabaseType::H2, Some("test")), "127.0.0.1", 9092, "test");
assert_eq!(params["sessionRole"], "workload");
assert_eq!(params["database_type"], "h2");
}
#[test]
+1
View File
@@ -172,6 +172,7 @@ macro_rules! agent_connection_pool_database_type {
| DatabaseType::Snowflake
| DatabaseType::Trino
| DatabaseType::Hive
| DatabaseType::Impala
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
+12 -1
View File
@@ -645,7 +645,7 @@ pub fn build_duplicate_table_structure_sql(options: DuplicateTableStructureSqlOp
let source = qualified_name(options.database_type, options.schema.as_deref(), &options.source_name);
let target =
qualified_duplicate_target_name(options.database_type, options.schema.as_deref(), &options.target_name);
let structure_sql = if options.database_type == Some(DatabaseType::Mysql) {
let structure_sql = if matches!(options.database_type, Some(DatabaseType::Mysql | DatabaseType::Impala)) {
format!("CREATE TABLE {target} LIKE {source};")
} else if options.database_type == Some(DatabaseType::Questdb) {
format!("CREATE TABLE {target} (LIKE {source});")
@@ -1684,6 +1684,17 @@ mod tests {
}),
"CREATE TABLE \"public\".\"users_copy\" (LIKE \"public\".\"users\" INCLUDING ALL);"
);
assert_eq!(
build_duplicate_table_structure_sql(DuplicateTableStructureSqlOptions {
database_type: Some(DatabaseType::Impala),
schema: Some("dbx_demo".to_string()),
source_name: "connection_test".to_string(),
target_name: "connection_test_copy".to_string(),
table_comment: None,
column_comments: vec![],
}),
"CREATE TABLE `dbx_demo`.`connection_test_copy` LIKE `dbx_demo`.`connection_test`;"
);
assert_eq!(
build_duplicate_table_structure_sql(DuplicateTableStructureSqlOptions {
database_type: Some(DatabaseType::Postgres),
+28
View File
@@ -564,6 +564,7 @@ pub enum DatabaseType {
#[serde(rename = "prestosql")]
PrestoSql,
Hive,
Impala,
Spark,
#[serde(rename = "db2")]
Db2,
@@ -1141,6 +1142,7 @@ impl ConnectionConfig {
DatabaseType::Trino => format!("trino://{host}:{port}{db_part}"),
DatabaseType::PrestoSql => format!("prestosql://{host}:{port}{db_part}"),
DatabaseType::Hive => format!("hive://{host}:{port}{db_part}"),
DatabaseType::Impala => format!("impala://{host}:{port}{db_part}"),
DatabaseType::Spark => format!("spark://{host}:{port}{db_part}"),
DatabaseType::Db2 => format!("db2://{host}:{port}{db_part}"),
DatabaseType::Informix => format!("informix://{host}:{port}{db_part}"),
@@ -1351,6 +1353,13 @@ impl ConnectionConfig {
DatabaseType::Hive => {
format!("hive://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Impala => {
if self.username.is_empty() && self.password.is_empty() {
format!("impala://{host}:{port}{db_part}")
} else {
format!("impala://{}:{}@{host}:{port}{db_part}", username, password)
}
}
DatabaseType::Spark => {
format!("spark://{}:{}@{host}:{port}{db_part}", username, password)
}
@@ -2599,6 +2608,25 @@ mod tests {
assert_eq!(serde_json::from_str::<DatabaseType>("\"easysearch\"").unwrap(), DatabaseType::Easysearch);
}
#[test]
fn impala_database_type_and_connection_urls_are_stable() {
assert_eq!(serde_json::to_string(&DatabaseType::Impala).unwrap(), "\"impala\"");
assert_eq!(serde_json::from_str::<DatabaseType>("\"impala\"").unwrap(), DatabaseType::Impala);
let mut config = mysql_config("", "", Some("analytics"));
config.db_type = DatabaseType::Impala;
config.host = "impala.local".to_string();
config.port = 21050;
assert_eq!(config.connection_url(), "impala://impala.local:21050/analytics");
config.username = "analyst".to_string();
config.password = "secret".to_string();
assert_eq!(
config.connection_url_with_host(&config.host, config.port),
"impala://analyst:secret@impala.local:21050/analytics"
);
}
#[test]
fn cloudflare_d1_database_type_and_api_url_are_stable() {
assert_eq!(serde_json::to_string(&DatabaseType::CloudflareD1).unwrap(), "\"cloudflare-d1\"");
@@ -53,6 +53,7 @@ pub fn is_schema_aware(database_type: DatabaseType) -> bool {
| DatabaseType::Trino
| DatabaseType::PrestoSql
| DatabaseType::Hive
| DatabaseType::Impala
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
@@ -669,8 +669,8 @@ pub fn profile_for(db_type: DatabaseType) -> DdlDialectProfile {
ClickHouse => clickhouse_profile(),
DuckDb | Questdb | SapHana | Teradata | Snowflake | Trino | PrestoSql | Hive | Spark | Db2 | Informix
| Bigquery | Kylin | Oscar | Tdengine | Iotdb | Databricks | Jdbc => conservative_ansi(db_type),
DuckDb | Questdb | SapHana | Teradata | Snowflake | Trino | PrestoSql | Hive | Impala | Spark | Db2
| Informix | Bigquery | Kylin | Oscar | Tdengine | Iotdb | Databricks | Jdbc => conservative_ansi(db_type),
// Non-tabular / not applicable for relational CREATE TABLE
Redis | MongoDb | Elasticsearch | Easysearch | Meilisearch | Qdrant | Milvus | Weaviate | ChromaDb | Neo4j
@@ -129,6 +129,7 @@ pub fn quote_table_identifier(database_type: Option<DatabaseType>, name: &str) -
| DatabaseType::StarRocks
| DatabaseType::ManticoreSearch
| DatabaseType::Hive
| DatabaseType::Impala
| DatabaseType::Spark
| DatabaseType::Databend
| DatabaseType::Tdengine
@@ -378,6 +379,7 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType
| DatabaseType::Doris
| DatabaseType::StarRocks
| DatabaseType::Hive
| DatabaseType::Impala
| DatabaseType::Spark
| DatabaseType::Questdb => format!("`{}`", name.replace('`', "``")),
DatabaseType::SqlServer => format!("[{}]", name.replace(']', "]]")),
@@ -186,6 +186,11 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
let default_order_by = if database_type == Some(DatabaseType::InfluxDb) {
// InfluxQL only allows sorting of timestamp column
Some("time DESC".to_string())
} else if database_type == Some(DatabaseType::Impala) {
// Impala requires ORDER BY when OFFSET is present. Keeping the same
// fallback on the first page also prevents page boundaries from using
// different row orders when the table has no explicit key.
Some("1".to_string())
} else {
None
};
@@ -502,7 +507,7 @@ pub(super) fn build_select_columns(
.collect::<Vec<_>>()
.join(", ");
}
if database_type != Some(DatabaseType::Hive) {
if !matches!(database_type, Some(DatabaseType::Hive | DatabaseType::Impala)) {
return "*".to_string();
}
columns
+31
View File
@@ -4,11 +4,13 @@ use crate::models::connection::DatabaseType;
#[test]
fn transfer_identifier_policy_preserves_legacy_output() {
assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::Hive), "`user``events`");
assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::Impala), "`user``events`");
assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::ClickHouse), "`user``events`");
assert_eq!(quote_transfer_identifier("user`events", &DatabaseType::Doris), "`user``events`");
assert_eq!(quote_transfer_identifier("user]events", &DatabaseType::SqlServer), "[user]]events]");
assert_eq!(quote_transfer_identifier("user\"events", &DatabaseType::Postgres), "\"user\"\"events\"");
assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Hive, None), "`warehouse`.`events`");
assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Impala, None), "`warehouse`.`events`");
assert_eq!(qualified_transfer_table("events", "warehouse", &DatabaseType::Mysql, None), "`events`");
}
@@ -245,6 +247,17 @@ fn builds_select_sql_with_limit_syntax_for_database_type() {
}),
"SELECT * FROM `test`.`dws_event_analyse` LIMIT 100;"
);
assert_eq!(
build_table_select_sql(TableSelectSqlOptions {
database_type: Some(DatabaseType::Impala),
schema: Some("dbx_demo"),
table_name: "connection_test",
columns: &[],
order_columns: &[],
limit: 100,
}),
"SELECT * FROM `dbx_demo`.`connection_test` LIMIT 100;"
);
assert_eq!(
build_table_select_sql(TableSelectSqlOptions {
database_type: Some(DatabaseType::StarRocks),
@@ -340,6 +353,24 @@ fn builds_table_data_where_and_schema_queries() {
}),
"SELECT * FROM \"public\".\"orders\" WHERE (amount > 10) LIMIT 50 OFFSET 100;"
);
assert_eq!(
build_table_data_select_sql(TableDataSelectSqlOptions {
database_type: Some(DatabaseType::Impala),
schema: Some("dbx_demo".to_string()),
table_name: "connection_test".to_string(),
table_type: Some("TABLE".to_string()),
primary_keys: Vec::new(),
columns: vec!["id".to_string(), "name".to_string()],
fallback_order_columns: Vec::new(),
order_by: None,
limit: Some(2),
offset: Some(1),
where_input: None,
include_row_id: false,
..Default::default()
}),
"SELECT `id` AS `id`, `name` AS `name` FROM `dbx_demo`.`connection_test` ORDER BY 1 LIMIT 2 OFFSET 1;"
);
assert_eq!(
build_table_data_select_sql(TableDataSelectSqlOptions {
database_type: Some(DatabaseType::Kingbase),
+281 -80
View File
@@ -5,10 +5,13 @@ use tokio::sync::RwLock;
use crate::connection::{config_for_pool_key, AppState, PoolKind};
use crate::db;
use crate::db::agent_driver::AgentTableReadStartParams;
use crate::db::mongo_driver::MongoDocumentResult;
use crate::models::connection::DatabaseType;
use crate::object_source_sql::{build_executable_object_source_statements, EditableObjectSourceSqlInput};
use crate::query::{agent_execute_query_params, pool_error_action, PoolErrorAction, QueryExecutionOptions};
use crate::query::{
agent_execute_query_params, pool_error_action, PoolErrorAction, QueryExecutionOptions, AGENT_PROTOCOL_MAX_ROWS,
};
use crate::sql::{split_sql_statements, split_sql_statements_for_database};
use crate::sql_dialect::{qualified_transfer_table, quote_transfer_identifier};
@@ -2652,7 +2655,7 @@ pub fn map_column_type(source_type: &str, _source_db: &DatabaseType, target_db:
// Extract basic type, `bigint unsigned` -> `bigint`
base = base.split(' ').next().unwrap_or(base).trim();
if matches!(target_db, DatabaseType::Hive) {
if matches!(target_db, DatabaseType::Hive | DatabaseType::Impala) {
return match base {
"tinyint" => "TINYINT".into(),
"smallint" | "int2" => "SMALLINT".into(),
@@ -2673,7 +2676,11 @@ pub fn map_column_type(source_type: &str, _source_db: &DatabaseType, target_db:
"TIMESTAMP".into()
}
"binary" | "varbinary" | "blob" | "tinyblob" | "mediumblob" | "longblob" | "bytea" | "image" => {
"BINARY".into()
if matches!(target_db, DatabaseType::Impala) {
"STRING".into()
} else {
"BINARY".into()
}
}
_ => "STRING".into(),
};
@@ -2852,7 +2859,7 @@ pub fn generate_create_table_ddl(
line.push(' ');
line.push_str(&default_clause);
}
if !c.is_nullable && !matches!(target_db, DatabaseType::Hive) {
if !c.is_nullable && !matches!(target_db, DatabaseType::Hive | DatabaseType::Impala) {
line.push_str(" NOT NULL");
}
if is_mysql_family {
@@ -2875,7 +2882,7 @@ pub fn generate_create_table_ddl(
}
let mut pks = Vec::with_capacity(columns.iter().filter(|c| c.is_primary_key).count());
if !matches!(target_db, DatabaseType::Hive) {
if !matches!(target_db, DatabaseType::Hive | DatabaseType::Impala) {
for c in columns {
if c.is_primary_key {
let qname = quote_identifier(&c.name, target_db);
@@ -3313,7 +3320,7 @@ fn generate_upsert_typed_for_transfer(
fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize {
match (db_type, mode) {
(DatabaseType::SqlServer, TransferMode::Append | TransferMode::Overwrite) => MAX_SQLSERVER_INSERT_ROWS,
(DatabaseType::Hive, _) => 500,
(DatabaseType::Hive | DatabaseType::Impala, _) => 500,
(DatabaseType::Oracle, TransferMode::Append | TransferMode::Overwrite) => MAX_ORACLE_INSERT_ALL_ROWS,
(DatabaseType::Oracle, TransferMode::Upsert) => MAX_ORACLE_MERGE_ROWS,
_ => usize::MAX,
@@ -6323,6 +6330,92 @@ where
Ok(total_transferred)
}
#[derive(Default)]
struct ImpalaTransferCursor {
started: bool,
session_id: Option<String>,
}
fn transfer_cursor_sql(
columns: &[String],
table: &str,
schema: &str,
db_type: &DatabaseType,
catalog: Option<&str>,
) -> String {
let full_table = qualified_table(table, schema, db_type, catalog);
let col_list = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
format!("SELECT {col_list} FROM {full_table}")
}
async fn fetch_impala_transfer_batch(
state: &AppState,
pool_key: &str,
request: &TransferRequest,
sql: &str,
batch_size: usize,
cursor: &mut ImpalaTransferCursor,
) -> Result<db::QueryResult, String> {
let query_timeout_secs = if cursor.started {
0
} else {
let configs = state.configs.read().await;
configs.get(&request.source_connection_id).map(|config| config.query_timeout_secs).unwrap_or(0)
};
let connections = state.connections.read().await;
let Some(PoolKind::Agent(client)) = connections.get(pool_key) else {
return Err("Impala transfer requires an Agent connection".to_string());
};
let client = client.clone();
drop(connections);
let mut client = client.lock().await;
let result = if cursor.started {
let session_id =
cursor.session_id.as_deref().ok_or("Impala transfer cursor ended before the next page was requested")?;
client.fetch_table_read_page::<db::QueryResult>(session_id, batch_size).await?
} else {
cursor.started = true;
client
.start_table_read::<db::QueryResult>(AgentTableReadStartParams {
sql: sql.to_string(),
database: Some(request.source_database.clone()),
schema: (!request.source_schema.trim().is_empty()).then(|| request.source_schema.clone()),
page_size: batch_size,
max_rows: AGENT_PROTOCOL_MAX_ROWS,
fetch_size: Some(batch_size),
timeout_secs: (query_timeout_secs > 0).then_some(query_timeout_secs),
})
.await?
};
if result.has_more {
cursor.session_id = result.session_id.clone().or_else(|| cursor.session_id.clone());
if cursor.session_id.is_none() {
return Err("Impala transfer cursor did not return a session id for additional rows".to_string());
}
} else {
cursor.session_id = None;
}
Ok(result)
}
async fn close_impala_transfer_cursor(state: &AppState, pool_key: &str, cursor: &mut ImpalaTransferCursor) {
let Some(session_id) = cursor.session_id.take() else {
return;
};
let connections = state.connections.read().await;
let Some(PoolKind::Agent(client)) = connections.get(pool_key) else {
return;
};
let client = client.clone();
drop(connections);
let mut client = client.lock().await;
if let Err(error) = client.close_table_read_session::<bool>(&session_id).await {
log::warn!("[transfer] failed to close Impala transfer cursor: {error}");
}
}
/// Transfer a single table. Returns rows transferred.
/// `progress_callback` is invoked for progress updates.
#[allow(clippy::too_many_arguments)]
@@ -6683,7 +6776,7 @@ where
}
let target_columns = if (request.mode == TransferMode::Upsert
&& !matches!(target_db_type, DatabaseType::ClickHouse | DatabaseType::Hive))
&& !matches!(target_db_type, DatabaseType::ClickHouse | DatabaseType::Hive | DatabaseType::Impala))
|| matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng)
{
get_columns_for_transfer(
@@ -6703,7 +6796,7 @@ where
// Determine effective mode and PK columns for upsert
let (effective_mode, pk_columns) = if request.mode == TransferMode::Upsert {
if matches!(target_db_type, DatabaseType::ClickHouse | DatabaseType::Hive) {
if matches!(target_db_type, DatabaseType::ClickHouse | DatabaseType::Hive | DatabaseType::Impala) {
log::warn!("[transfer] upsert not supported for {:?}, falling back to append", target_db_type);
(TransferMode::Append, vec![])
} else {
@@ -6730,92 +6823,120 @@ where
let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size };
let mut offset: u64 = 0;
let mut total_transferred: u64 = 0;
loop {
if is_cancelled(&request.transfer_id).await {
return Err("Cancelled".to_string());
}
let sql = pagination_sql_with_order(
// A single Agent cursor keeps Impala rows in one query execution. Re-running
// LIMIT/OFFSET pages cannot be made stable for tables without a unique key.
let use_impala_cursor = matches!(source_db_type, DatabaseType::Impala);
let impala_transfer_sql = use_impala_cursor.then(|| {
transfer_cursor_sql(
&col_names,
table,
&request.source_schema,
source_db_type,
offset,
batch_size,
&primary_key_columns,
request.source_catalog.as_deref(),
);
let (sql, mysql_spatial_markers) =
mysql_spatial_transfer_select_sql(sql, &col_names, &col_types, source_db_type, target_db_type);
let result = execute_on_pool(state, source_pool_key, &sql).await?;
let row_count = result.rows.len();
)
});
let mut impala_cursor = ImpalaTransferCursor::default();
if row_count == 0 {
break;
}
let transfer_result: Result<(), String> = async {
loop {
if is_cancelled(&request.transfer_id).await {
return Err("Cancelled".to_string());
}
let write_statements = generate_transfer_write_sql_batches(
&effective_mode,
&col_names,
&col_types,
&result.rows,
&target_table,
&request.target_schema,
target_db_type,
&pk_columns,
request.target_catalog.as_deref(),
overrides_postgres_system_values,
mysql_spatial_markers,
)?;
for (statement_index, batch_sql) in write_statements.iter().enumerate() {
execute_transfer_write_statement(
state,
target_pool_key,
batch_sql,
target_db_type,
let (result, mysql_spatial_markers) = if let Some(sql) = impala_transfer_sql.as_deref() {
(
fetch_impala_transfer_batch(state, source_pool_key, request, sql, batch_size, &mut impala_cursor)
.await?,
false,
)
} else {
let sql = pagination_sql_with_order(
&col_names,
table,
&request.source_schema,
source_db_type,
offset,
batch_size,
&primary_key_columns,
request.source_catalog.as_deref(),
);
let (sql, mysql_spatial_markers) =
mysql_spatial_transfer_select_sql(sql, &col_names, &col_types, source_db_type, target_db_type);
(execute_on_pool(state, source_pool_key, &sql).await?, mysql_spatial_markers)
};
let has_more = result.has_more;
let row_count = result.rows.len();
if row_count == 0 {
break;
}
let write_statements = generate_transfer_write_sql_batches(
&effective_mode,
&col_names,
&col_types,
&result.rows,
&target_table,
&request.target_schema,
writes_dameng_identity_columns,
)
.await
.map_err(|e| {
let absolute_row = parse_mysql_row_error(&e).map(|row| offset + row);
match absolute_row {
Some(row) => format!(
"Insert failed for table '{target_table}' at row {row} (chunk {} of {}): {e}",
statement_index + 1,
write_statements.len()
),
None => format!(
"Insert failed for table '{target_table}' at offset {offset}, chunk {} of {}: {e}",
statement_index + 1,
write_statements.len()
),
}
})?;
}
target_db_type,
&pk_columns,
request.target_catalog.as_deref(),
overrides_postgres_system_values,
mysql_spatial_markers,
)?;
for (statement_index, batch_sql) in write_statements.iter().enumerate() {
execute_transfer_write_statement(
state,
target_pool_key,
batch_sql,
target_db_type,
&target_table,
&request.target_schema,
writes_dameng_identity_columns,
)
.await
.map_err(|e| {
let absolute_row = parse_mysql_row_error(&e).map(|row| offset + row);
match absolute_row {
Some(row) => format!(
"Insert failed for table '{target_table}' at row {row} (chunk {} of {}): {e}",
statement_index + 1,
write_statements.len()
),
None => format!(
"Insert failed for table '{target_table}' at offset {offset}, chunk {} of {}: {e}",
statement_index + 1,
write_statements.len()
),
}
})?;
}
total_transferred += row_count as u64;
log::info!("[transfer] {} batch +{} rows (total {})", table, row_count, total_transferred);
offset += row_count as u64;
total_transferred += row_count as u64;
log::info!("[transfer] {} batch +{} rows (total {})", table, row_count, total_transferred);
offset += row_count as u64;
progress_callback(TransferProgress {
transfer_id: request.transfer_id.clone(),
table: table.to_string(),
table_index,
total_tables,
rows_transferred: total_transferred,
total_rows,
status: TransferStatus::Running,
error: None,
terminal: false,
});
progress_callback(TransferProgress {
transfer_id: request.transfer_id.clone(),
table: table.to_string(),
table_index,
total_tables,
rows_transferred: total_transferred,
total_rows,
status: TransferStatus::Running,
error: None,
terminal: false,
});
if row_count < batch_size {
break;
if (use_impala_cursor && !has_more) || (!use_impala_cursor && row_count < batch_size) {
break;
}
}
Ok(())
}
.await;
close_impala_transfer_cursor(state, source_pool_key, &mut impala_cursor).await;
transfer_result?;
if pg_compat_transfer {
for statement in generate_postgres_sequence_sync_sql(&columns, &target_table, &request.target_schema) {
@@ -9214,6 +9335,54 @@ mod tests {
);
}
#[test]
fn impala_transfer_uses_impala_compatible_ddl_and_batches() {
let cols = vec![
db::ColumnInfo { is_primary_key: true, is_nullable: false, ..test_column("id", "bigint") },
db::ColumnInfo { is_nullable: false, ..test_column("payload", "jsonb") },
db::ColumnInfo { is_nullable: true, ..test_column("binary_payload", "bytea") },
];
let ddl = generate_create_table_ddl(
&cols,
"events",
"public",
"warehouse",
&DatabaseType::Impala,
&DatabaseType::Postgres,
None,
None,
);
assert!(ddl.contains("CREATE TABLE IF NOT EXISTS `warehouse`.`events`"));
assert!(ddl.contains("`id` BIGINT"));
assert!(ddl.contains("`payload` STRING"));
assert!(ddl.contains("`binary_payload` STRING"));
assert!(!ddl.contains("PRIMARY KEY"));
assert!(!ddl.contains("NOT NULL"));
assert_eq!(map_column_type("varbinary(255)", &DatabaseType::Mysql, &DatabaseType::Impala), "STRING");
assert_eq!(map_column_type("bytea", &DatabaseType::Postgres, &DatabaseType::Impala), "STRING");
assert_eq!(map_column_type("binary", &DatabaseType::Mysql, &DatabaseType::Hive), "BINARY");
let statements = generate_transfer_write_sql_batches(
&TransferMode::Append,
&[String::from("id"), String::from("binary_payload")],
&[Some(String::from("bigint")), Some(String::from("bytea"))],
&[vec![json!(1), json!("0x00ff")]],
"events",
"warehouse",
&DatabaseType::Impala,
&[],
None,
false,
false,
)
.unwrap();
assert_eq!(statements, vec!["INSERT INTO `warehouse`.`events` (`id`, `binary_payload`) VALUES\n(1, '0x00ff')"]);
assert_eq!(max_transfer_write_rows(&DatabaseType::Impala, &TransferMode::Append), 500);
assert_eq!(max_transfer_write_rows(&DatabaseType::Impala, &TransferMode::Upsert), 500);
}
#[test]
fn mongo_transfer_document_fields_preserve_first_seen_order() {
let documents = vec![json!({"b": 1}), json!({"a": 2, "c": 3}), json!({"b": 4, "d": 5})];
@@ -9260,6 +9429,38 @@ mod tests {
assert_eq!(sql, "SELECT \"id\", \"name\" FROM \"public\".\"users\" ORDER BY \"id\" LIMIT 100 OFFSET 200");
}
#[test]
fn impala_transfer_without_primary_key_uses_one_cursor_query() {
let sql = transfer_cursor_sql(
&[String::from("id"), String::from("name")],
"events",
"analytics",
&DatabaseType::Impala,
None,
);
assert_eq!(sql, "SELECT `id`, `name` FROM `analytics`.`events`");
assert!(!sql.contains("ORDER BY"));
assert!(!sql.contains("LIMIT"));
assert!(!sql.contains("OFFSET"));
}
#[test]
fn impala_transfer_preserves_explicit_primary_key_order() {
let sql = pagination_sql_with_order(
&[String::from("id"), String::from("name")],
"events",
"analytics",
&DatabaseType::Impala,
1000,
1000,
&[String::from("id")],
None,
);
assert_eq!(sql, "SELECT `id`, `name` FROM `analytics`.`events` ORDER BY `id` LIMIT 1000 OFFSET 1000");
}
#[test]
fn doris_unique_key_columns_drive_transfer_pagination_order() {
let columns =
@@ -97,6 +97,7 @@ fn driver_manifest() -> DriverManifest {
fn maps_agent_database_types_to_driver_keys() {
assert_eq!(agent_key(&DatabaseType::Trino, None), Some("trino"));
assert_eq!(agent_key(&DatabaseType::Hive, None), Some("hive"));
assert_eq!(agent_key(&DatabaseType::Impala, None), Some("hive"));
assert_eq!(agent_key(&DatabaseType::Tdengine, None), Some("tdengine"));
assert_eq!(agent_key(&DatabaseType::Iotdb, None), Some("iotdb"));
assert_eq!(agent_key(&DatabaseType::Yashandb, None), Some("yashandb"));
@@ -139,6 +140,7 @@ fn classifies_agent_database_types() {
assert!(is_agent_type(&DatabaseType::Oracle));
assert!(is_agent_type(&DatabaseType::Trino));
assert!(is_agent_type(&DatabaseType::Hive));
assert!(is_agent_type(&DatabaseType::Impala));
assert!(is_agent_type(&DatabaseType::Tdengine));
assert!(is_agent_type(&DatabaseType::Iotdb));
assert!(is_agent_type(&DatabaseType::Yashandb));
@@ -352,6 +354,25 @@ fn driver_manifest_declares_expected_product_capabilities() {
let starrocks = find_driver(DatabaseType::StarRocks);
assert!(starrocks.capabilities.user_admin);
let impala = find_driver(DatabaseType::Impala);
assert_eq!(impala.label, "Apache Impala");
assert_eq!(impala.runtime_mode, "agent");
assert_eq!(impala.agent_key.as_deref(), Some("hive"));
assert_eq!(impala.support_level, "operate");
assert!(impala.capabilities.query_execution);
assert!(impala.capabilities.metadata_browse);
assert!(impala.capabilities.object_browser);
assert!(impala.capabilities.object_source);
assert!(impala.capabilities.schema_search);
assert!(impala.capabilities.sql_explain);
assert!(impala.capabilities.data_transfer);
assert!(impala.capabilities.sql_file_execution);
assert!(!impala.capabilities.table_data_edit);
assert!(!impala.capabilities.table_structure_edit);
assert!(!impala.capabilities.table_import);
assert!(!impala.capabilities.diagram);
assert!(!impala.capabilities.user_admin);
}
#[test]