Merge pull request #1580 from ydb-platform/xsql

database/sql over query service client

.github/workflows/examples.yml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        ydb-version: [ 23.3, 24.1 ]
+        ydb-version: [ 24.1, 24.2, 24.3 ]
         application: [ native/table, native/query, database_sql, gorm, xorm ]
     services:
       ydb:

.github/workflows/tests.yml

@@ -19,7 +19,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go-version: [1.21.x, 1.22.x, 1.23.x]
+        go-version: [1.22.x, 1.23.x]
         os: [ubuntu, windows, macOS]
     env:
       OS: ${{ matrix.os }}-latest
@@ -52,8 +52,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        go-version: [1.21.x, 1.22.x, 1.23.x]
-        ydb-version: [23.3, 24.1, 24.2]
+        go-version: [1.22.x, 1.23.x]
+        ydb-version: [24.1, 24.2, 24.3]
         os: [ubuntu]
     services:
       ydb:
@@ -133,6 +133,7 @@ jobs:
       YDB_CONNECTION_STRING_SECURE: grpcs://localhost:2135/local
       YDB_SSL_ROOT_CERTIFICATES_FILE: /tmp/ydb_certs/ca.pem
       YDB_SESSIONS_SHUTDOWN_URLS: http://localhost:8765/actors/kqp_proxy?force_shutdown=all
+      YDB_DATABASE_SQL_OVER_QUERY_SERVICE: 1
       HIDE_APPLICATION_OUTPUT: 1
     steps:
       - name: Checkout code

CHANGELOG.md

@@ -1,3 +1,7 @@
+* Added implementation of `database/sql` driver over query service client
+* Added `ydb.WithQueryService(bool)` option to explicitly enable `database/sql` driver over query service client
+* Added environment parameter `YDB_DATABASE_SQL_OVER_QUERY_SERVICE` to enable `database/sql` driver over query service client without code rewriting
+
 ## v3.94.0
 * Refactored golang types mapping into ydb types using `ydb.ParamsFromMap` and `database/sql` query arguments
 * Small breaking change: type mapping for `ydb.ParamsFromMap` and `database/sql` type `uuid.UUID` changed from ydb type `Text` to ydb type `UUID`

driver.go

@@ -14,7 +14,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/discovery"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/balancer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/conn"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/connector"
 	internalCoordination "github.com/ydb-platform/ydb-go-sdk/v3/internal/coordination"
 	coordinationConfig "github.com/ydb-platform/ydb-go-sdk/v3/internal/coordination/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/credentials"
@@ -37,6 +36,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicclientinternal"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsync"
 	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"github.com/ydb-platform/ydb-go-sdk/v3/operation"
@@ -93,7 +93,7 @@ type (
 		topic        *xsync.Once[*topicclientinternal.Client]
 		topicOptions []topicoptions.TopicOption
 
-		databaseSQLOptions []connector.Option
+		databaseSQLOptions []xsql.Option
 
 		pool *conn.Pool
 

dsn.go

@@ -9,10 +9,9 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/balancers"
 	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/bind"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/connector"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/dsn"
-	tableSql "github.com/ydb-platform/ydb-go-sdk/v3/internal/table/conn"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
 )
 
 const tablePathPrefixTransformer = "table_path_prefix"
@@ -44,6 +43,22 @@ func UnregisterDsnParser(registrationID int) {
 	dsnParsers[registrationID] = nil
 }
 
+var stringToType = map[string]QueryMode{
+	"data":      DataQueryMode,
+	"scan":      ScanQueryMode,
+	"scheme":    SchemeQueryMode,
+	"scripting": ScriptingQueryMode,
+	"query":     QueryExecuteQueryMode,
+}
+
+func queryModeFromString(s string) QueryMode {
+	if t, ok := stringToType[s]; ok {
+		return t
+	}
+
+	return unknownQueryMode
+}
+
 //nolint:funlen
 func parseConnectionString(dataSourceName string) (opts []Option, _ error) {
 	info, err := dsn.Parse(dataSourceName)
@@ -60,45 +75,52 @@ func parseConnectionString(dataSourceName string) (opts []Option, _ error) {
 		opts = append(opts, WithBalancer(balancers.FromConfig(balancer)))
 	}
 	if queryMode := info.Params.Get("go_query_mode"); queryMode != "" {
-		mode := tableSql.QueryModeFromString(queryMode)
-		if mode == tableSql.UnknownQueryMode {
+		switch mode := queryModeFromString(queryMode); mode {
+		case QueryExecuteQueryMode:
+			opts = append(opts, withConnectorOptions(xsql.WithQueryService(true)))
+		case unknownQueryMode:
 			return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
+		default:
+			opts = append(opts, withConnectorOptions(xsql.WithDefaultQueryMode(modeToMode(mode))))
 		}
-		opts = append(opts, withConnectorOptions(connector.WithDefaultQueryMode(mode)))
 	} else if queryMode := info.Params.Get("query_mode"); queryMode != "" {
-		mode := tableSql.QueryModeFromString(queryMode)
-		if mode == tableSql.UnknownQueryMode {
+		switch mode := queryModeFromString(queryMode); mode {
+		case QueryExecuteQueryMode:
+			opts = append(opts, withConnectorOptions(xsql.WithQueryService(true)))
+		case unknownQueryMode:
 			return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
+		default:
+			opts = append(opts, withConnectorOptions(xsql.WithDefaultQueryMode(modeToMode(mode))))
 		}
-		opts = append(opts, withConnectorOptions(connector.WithDefaultQueryMode(mode)))
 	}
 	if fakeTx := info.Params.Get("go_fake_tx"); fakeTx != "" {
 		for _, queryMode := range strings.Split(fakeTx, ",") {
-			mode := tableSql.QueryModeFromString(queryMode)
-			if mode == tableSql.UnknownQueryMode {
+			switch mode := queryModeFromString(queryMode); mode {
+			case unknownQueryMode:
 				return nil, xerrors.WithStackTrace(fmt.Errorf("unknown query mode: %s", queryMode))
+			default:
+				opts = append(opts, withConnectorOptions(WithFakeTx(mode)))
 			}
-			opts = append(opts, withConnectorOptions(connector.WithFakeTx(mode)))
 		}
 	}
 	if info.Params.Has("go_query_bind") {
-		var binders []connector.Option
+		var binders []xsql.Option
 		queryTransformers := strings.Split(info.Params.Get("go_query_bind"), ",")
 		for _, transformer := range queryTransformers {
 			switch transformer {
 			case "declare":
-				binders = append(binders, connector.WithQueryBind(bind.AutoDeclare{}))
+				binders = append(binders, xsql.WithQueryBind(bind.AutoDeclare{}))
 			case "positional":
-				binders = append(binders, connector.WithQueryBind(bind.PositionalArgs{}))
+				binders = append(binders, xsql.WithQueryBind(bind.PositionalArgs{}))
 			case "numeric":
-				binders = append(binders, connector.WithQueryBind(bind.NumericArgs{}))
+				binders = append(binders, xsql.WithQueryBind(bind.NumericArgs{}))
 			default:
 				if strings.HasPrefix(transformer, tablePathPrefixTransformer) {
 					prefix, err := extractTablePathPrefixFromBinderName(transformer)
 					if err != nil {
 						return nil, xerrors.WithStackTrace(err)
 					}
-					binders = append(binders, connector.WithQueryBind(bind.TablePathPrefix(prefix)))
+					binders = append(binders, xsql.WithQueryBind(bind.TablePathPrefix(prefix)))
 				} else {
 					return nil, xerrors.WithStackTrace(
 						fmt.Errorf("unknown query rewriter: %s", transformer),

dsn_test.go

@@ -8,14 +8,14 @@ import (
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/bind"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/connector"
-	querySql "github.com/ydb-platform/ydb-go-sdk/v3/internal/query/conn"
-	tableSql "github.com/ydb-platform/ydb-go-sdk/v3/internal/table/conn"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql/legacy"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql/propose"
 )
 
 func TestParse(t *testing.T) {
-	newConnector := func(opts ...connector.Option) *connector.Connector {
-		c := &connector.Connector{}
+	newConnector := func(opts ...xsql.Option) *xsql.Connector {
+		c := &xsql.Connector{}
 		for _, opt := range opts {
 			if opt != nil {
 				if err := opt.Apply(c); err != nil {
@@ -26,11 +26,11 @@ func TestParse(t *testing.T) {
 
 		return c
 	}
-	newTableConn := func(opts ...tableSql.Option) *tableSql.Conn {
-		return tableSql.New(context.Background(), nil, nil, opts...)
+	newLegacyConn := func(opts ...legacy.Option) *legacy.Conn {
+		return legacy.New(context.Background(), nil, nil, opts...)
 	}
-	newQueryConn := func(opts ...querySql.Option) *querySql.Conn {
-		return querySql.New(context.Background(), nil, nil, opts...)
+	newQueryConn := func(opts ...propose.Option) *propose.Conn {
+		return propose.New(context.Background(), nil, nil, opts...)
 	}
 	compareConfigs := func(t *testing.T, lhs, rhs *config.Config) {
 		require.Equal(t, lhs.Secure(), rhs.Secure())
@@ -40,7 +40,7 @@ func TestParse(t *testing.T) {
 	for _, tt := range []struct {
 		dsn           string
 		opts          []config.Option
-		connectorOpts []connector.Option
+		connectorOpts []xsql.Option
 		err           error
 	}{
 		{
@@ -70,8 +70,8 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
 			},
 			err: nil,
 		},
@@ -82,9 +82,9 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
 			},
 			err: nil,
 		},
@@ -95,10 +95,10 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
-				connector.WithQueryBind(bind.NumericArgs{}),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+				xsql.WithQueryBind(bind.NumericArgs{}),
 			},
 			err: nil,
 		},
@@ -109,10 +109,10 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
-				connector.WithQueryBind(bind.PositionalArgs{}),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+				xsql.WithQueryBind(bind.PositionalArgs{}),
 			},
 			err: nil,
 		},
@@ -123,10 +123,10 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
-				connector.WithQueryBind(bind.AutoDeclare{}),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+				xsql.WithQueryBind(bind.AutoDeclare{}),
 			},
 			err: nil,
 		},
@@ -137,9 +137,9 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
 			},
 			err: nil,
 		},
@@ -150,11 +150,11 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithDefaultQueryMode(tableSql.ScriptingQueryMode),
-				connector.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
-				connector.WithQueryBind(bind.PositionalArgs{}),
-				connector.WithQueryBind(bind.AutoDeclare{}),
+			connectorOpts: []xsql.Option{
+				xsql.WithDefaultQueryMode(legacy.ScriptingQueryMode),
+				xsql.WithQueryBind(bind.TablePathPrefix("path/to/tables")),
+				xsql.WithQueryBind(bind.PositionalArgs{}),
+				xsql.WithQueryBind(bind.AutoDeclare{}),
 			},
 			err: nil,
 		},
@@ -165,9 +165,9 @@ func TestParse(t *testing.T) {
 				config.WithEndpoint("localhost:2135"),
 				config.WithDatabase("/local"),
 			},
-			connectorOpts: []connector.Option{
-				connector.WithFakeTx(tableSql.ScriptingQueryMode),
-				connector.WithFakeTx(tableSql.SchemeQueryMode),
+			connectorOpts: []xsql.Option{
+				WithFakeTx(ScriptingQueryMode),
+				WithFakeTx(SchemeQueryMode),
 			},
 			err: nil,
 		},
@@ -183,15 +183,15 @@ func TestParse(t *testing.T) {
 				exp := newConnector(tt.connectorOpts...)
 				act := newConnector(d.databaseSQLOptions...)
 				t.Run("tableOptions", func(t *testing.T) {
-					require.Equal(t, newTableConn(exp.TableOpts...), newTableConn(act.TableOpts...))
+					require.Equal(t, newLegacyConn(exp.LegacyOpts...), newLegacyConn(act.LegacyOpts...))
 				})
 				t.Run("queryOptions", func(t *testing.T) {
-					require.Equal(t, newQueryConn(exp.QueryOpts...), newQueryConn(act.QueryOpts...))
+					require.Equal(t, newQueryConn(exp.Options...), newQueryConn(act.Options...))
 				})
-				exp.TableOpts = nil
-				exp.QueryOpts = nil
-				act.TableOpts = nil
-				act.QueryOpts = nil
+				exp.LegacyOpts = nil
+				exp.Options = nil
+				act.LegacyOpts = nil
+				act.Options = nil
 				require.Equal(t, exp.Bindings(), act.Bindings())
 				require.Equal(t, exp, act)
 				compareConfigs(t, config.New(tt.opts...), d.config)

examples/ddl/ddl.go

@@ -67,10 +67,10 @@ ALTER TABLE small_table3 SET (TTL = Interval("PT3H") ON d);
 `
 )
 
-func executeQuery(ctx context.Context, c table.Client, prefix, query string) (err error) {
+func executeQuery(ctx context.Context, c table.Client, prefix, sql string) (err error) {
 	err = c.Do(ctx,
 		func(ctx context.Context, s table.Session) error {
-			err = s.ExecuteSchemeQuery(ctx, fmt.Sprintf(query, prefix))
+			err = s.ExecuteSchemeQuery(ctx, fmt.Sprintf(sql, prefix))
 
 			return err
 		},

internal/bind/auto_declare.go

@@ -13,7 +13,7 @@ func (m AutoDeclare) blockID() blockID {
 	return blockDeclare
 }
 
-func (m AutoDeclare) RewriteQuery(query string, args ...interface{}) (
+func (m AutoDeclare) ToYdb(sql string, args ...interface{}) (
 	yql string, newArgs []interface{}, err error,
 ) {
 	params, err := Params(args...)
@@ -22,7 +22,7 @@ func (m AutoDeclare) RewriteQuery(query string, args ...interface{}) (
 	}
 
 	if len(params) == 0 {
-		return query, args, nil
+		return sql, args, nil
 	}
 
 	var (
@@ -46,7 +46,7 @@ func (m AutoDeclare) RewriteQuery(query string, args ...interface{}) (
 
 	buffer.WriteByte('\n')
 
-	buffer.WriteString(query)
+	buffer.WriteString(sql)
 
 	for _, param := range params {
 		newArgs = append(newArgs, param)