YQ Connector:new interface of query construct

16 files changed, 131 insertions, 197 deletions

diff --git a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/query_executor.go b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/query_executor.go
deleted file mode 100644
index a3b469d58c..0000000000
--- a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/query_executor.go
+++ /dev/null
@@ -1,35 +0,0 @@
-package clickhouse
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/app/server/utils"
-	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
-)
-
-type queryExecutor struct {
-}
-
-func (qm queryExecutor) DescribeTable(ctx context.Context, conn utils.Connection, request *api_service_protos.TDescribeTableRequest) (utils.Rows, error) {
-	out, err := conn.Query(
-		ctx,
-		"SELECT name, type FROM system.columns WHERE table = ? and database = ?",
-		request.Table,
-		request.DataSourceInstance.Database,
-	)
-
-	if err != nil {
-		return nil, fmt.Errorf("query: %w", err)
-	}
-
-	if err := out.Err(); err != nil {
-		return nil, fmt.Errorf("rows err: %w", err)
-	}
-
-	return out, nil
-}
-
-func NewQueryExecutor() utils.QueryExecutor {
-	return queryExecutor{}
-}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter.go b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter.go
index 4f87ce1c3a..d2788a20d7 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter.go
@@ -1,22 +1,17 @@
 package clickhouse
 
 import (
-	"fmt"
-	"strings"
-
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
-	"github.com/ydb-platform/ydb/library/go/core/log"
 	"github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/app/server/utils"
 	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
 )
 
-type sqlFormatter struct {
-}
+var _ utils.SQLFormatter = (*sqlFormatter)(nil)
 
-type predicateBuilderFeatures struct {
+type sqlFormatter struct {
 }
 
-func (f predicateBuilderFeatures) SupportsType(typeID Ydb.Type_PrimitiveTypeId) bool {
+func (f *sqlFormatter) supportsType(typeID Ydb.Type_PrimitiveTypeId) bool {
 	switch typeID {
 	case Ydb.Type_BOOL:
 		return true
@@ -45,23 +40,23 @@ func (f predicateBuilderFeatures) SupportsType(typeID Ydb.Type_PrimitiveTypeId)
 	}
 }
 
-func (f predicateBuilderFeatures) SupportsConstantValueExpression(t *Ydb.Type) bool {
+func (f *sqlFormatter) supportsConstantValueExpression(t *Ydb.Type) bool {
 	switch v := t.Type.(type) {
 	case *Ydb.Type_TypeId:
-		return f.SupportsType(v.TypeId)
+		return f.supportsType(v.TypeId)
 	case *Ydb.Type_OptionalType:
-		return f.SupportsConstantValueExpression(v.OptionalType.Item)
+		return f.supportsConstantValueExpression(v.OptionalType.Item)
 	default:
 		return false
 	}
 }
 
-func (f predicateBuilderFeatures) SupportsExpression(expression *api_service_protos.TExpression) bool {
+func (f sqlFormatter) SupportsPushdownExpression(expression *api_service_protos.TExpression) bool {
 	switch e := expression.Payload.(type) {
 	case *api_service_protos.TExpression_Column:
 		return true
 	case *api_service_protos.TExpression_TypedValue:
-		return f.SupportsConstantValueExpression(e.TypedValue.Type)
+		return f.supportsConstantValueExpression(e.TypedValue.Type)
 	case *api_service_protos.TExpression_ArithmeticalExpression:
 		return false
 	case *api_service_protos.TExpression_Null:
@@ -71,31 +66,20 @@ func (f predicateBuilderFeatures) SupportsExpression(expression *api_service_pro
 	}
 }
 
-func (formatter sqlFormatter) FormatRead(logger log.Logger, selectReq *api_service_protos.TSelect) (string, error) {
-	var sb strings.Builder
-
-	selectPart, err := utils.FormatSelectColumns(selectReq.What, selectReq.GetFrom().GetTable(), true)
-	if err != nil {
-		return "", fmt.Errorf("failed to format select statement: %w", err)
-	}
+func (f sqlFormatter) GetDescribeTableQuery(request *api_service_protos.TDescribeTableRequest) (string, []any) {
+	query := "SELECT name, type FROM system.columns WHERE table = ? and database = ?"
+	args := []any{request.Table, request.DataSourceInstance.Database}
 
-	sb.WriteString(selectPart)
-
-	if selectReq.Where != nil {
-		var features predicateBuilderFeatures
-
-		clause, err := utils.FormatWhereClause(selectReq.Where, features)
-		if err != nil {
-			logger.Error("Failed to format WHERE clause", log.Error(err), log.String("where", selectReq.Where.String()))
-		} else {
-			sb.WriteString(" ")
-			sb.WriteString(clause)
-		}
-	}
+	return query, args
+}
 
-	query := sb.String()
+func (f sqlFormatter) GetPlaceholder(_ int) string {
+	return "?"
+}
 
-	return query, nil
+func (f sqlFormatter) SanitiseIdentifier(ident string) string {
+	// TODO: sanitise
+	return ident
 }
 
 func NewSQLFormatter() utils.SQLFormatter {
diff --git a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter_test.go b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter_test.go
index 5b6f264d98..05a78854de 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter_test.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/sql_formatter_test.go
@@ -330,7 +330,7 @@ func TestSQLFormatter(t *testing.T) {
 		tc := tc
 
 		t.Run(tc.testName, func(t *testing.T) {
-			output, err := formatter.FormatRead(logger, tc.selectReq)
+			output, _, err := utils.MakeReadSplitQuery(logger, formatter, tc.selectReq)
 			require.Equal(t, tc.output, output)
 
 			if tc.err != nil {
diff --git a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/ya.make b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/ya.make
index a08bd57be7..46287aade5 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/clickhouse/ya.make
+++ b/ydb/library/yql/providers/generic/connector/app/server/clickhouse/ya.make
@@ -3,7 +3,6 @@ GO_LIBRARY()
 SRCS(
     connection_manager.go
     doc.go
-    query_executor.go
     sql_formatter.go
     type_mapper.go
 )
diff --git a/ydb/library/yql/providers/generic/connector/app/server/postgresql/query_executor.go b/ydb/library/yql/providers/generic/connector/app/server/postgresql/query_executor.go
deleted file mode 100644
index e894bfb3c7..0000000000
--- a/ydb/library/yql/providers/generic/connector/app/server/postgresql/query_executor.go
+++ /dev/null
@@ -1,32 +0,0 @@
-package postgresql
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/app/server/utils"
-	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
-)
-
-type queryExecutor struct {
-}
-
-func (qm queryExecutor) DescribeTable(ctx context.Context, conn utils.Connection, request *api_service_protos.TDescribeTableRequest) (utils.Rows, error) {
-	schema := request.GetDataSourceInstance().GetPgOptions().GetSchema()
-	out, err := conn.Query(
-		ctx,
-		"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = $1 AND table_schema = $2",
-		request.Table,
-		schema,
-	)
-
-	if err != nil {
-		return nil, fmt.Errorf("connection query: %w", err)
-	}
-
-	return out, nil
-}
-
-func NewQueryExecutor() utils.QueryExecutor {
-	return queryExecutor{}
-}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter.go b/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter.go
index fcaad0520b..ef59c7b182 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter.go
@@ -2,21 +2,18 @@ package postgresql
 
 import (
 	"fmt"
-	"strings"
 
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
-	"github.com/ydb-platform/ydb/library/go/core/log"
 	"github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/app/server/utils"
 	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
 )
 
-type sqlFormatter struct {
-}
+var _ utils.SQLFormatter = (*sqlFormatter)(nil)
 
-type predicateBuilderFeatures struct {
+type sqlFormatter struct {
 }
 
-func (f predicateBuilderFeatures) SupportsType(typeID Ydb.Type_PrimitiveTypeId) bool {
+func (f *sqlFormatter) supportsType(typeID Ydb.Type_PrimitiveTypeId) bool {
 	switch typeID {
 	case Ydb.Type_BOOL:
 		return true
@@ -45,23 +42,23 @@ func (f predicateBuilderFeatures) SupportsType(typeID Ydb.Type_PrimitiveTypeId)
 	}
 }
 
-func (f predicateBuilderFeatures) SupportsConstantValueExpression(t *Ydb.Type) bool {
+func (f *sqlFormatter) supportsConstantValueExpression(t *Ydb.Type) bool {
 	switch v := t.Type.(type) {
 	case *Ydb.Type_TypeId:
-		return f.SupportsType(v.TypeId)
+		return f.supportsType(v.TypeId)
 	case *Ydb.Type_OptionalType:
-		return f.SupportsConstantValueExpression(v.OptionalType.Item)
+		return f.supportsConstantValueExpression(v.OptionalType.Item)
 	default:
 		return false
 	}
 }
 
-func (f predicateBuilderFeatures) SupportsExpression(expression *api_service_protos.TExpression) bool {
+func (f sqlFormatter) SupportsPushdownExpression(expression *api_service_protos.TExpression) bool {
 	switch e := expression.Payload.(type) {
 	case *api_service_protos.TExpression_Column:
 		return true
 	case *api_service_protos.TExpression_TypedValue:
-		return f.SupportsConstantValueExpression(e.TypedValue.Type)
+		return f.supportsConstantValueExpression(e.TypedValue.Type)
 	case *api_service_protos.TExpression_ArithmeticalExpression:
 		return false
 	case *api_service_protos.TExpression_Null:
@@ -71,31 +68,21 @@ func (f predicateBuilderFeatures) SupportsExpression(expression *api_service_pro
 	}
 }
 
-func (formatter sqlFormatter) FormatRead(logger log.Logger, selectReq *api_service_protos.TSelect) (string, error) {
-	var sb strings.Builder
-
-	selectPart, err := utils.FormatSelectColumns(selectReq.What, selectReq.GetFrom().GetTable(), true)
-	if err != nil {
-		return "", fmt.Errorf("failed to format select statement: %w", err)
-	}
-
-	sb.WriteString(selectPart)
-
-	if selectReq.Where != nil {
-		var features predicateBuilderFeatures
+func (f sqlFormatter) GetDescribeTableQuery(request *api_service_protos.TDescribeTableRequest) (string, []any) {
+	schema := request.GetDataSourceInstance().GetPgOptions().GetSchema()
+	query := "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = $1 AND table_schema = $2"
+	args := []any{request.Table, schema}
 
-		clause, err := utils.FormatWhereClause(selectReq.Where, features)
-		if err != nil {
-			logger.Error("Failed to format WHERE clause", log.Error(err), log.String("where", selectReq.Where.String()))
-		} else {
-			sb.WriteString(" ")
-			sb.WriteString(clause)
-		}
-	}
+	return query, args
+}
 
-	query := sb.String()
+func (f sqlFormatter) GetPlaceholder(n int) string {
+	return fmt.Sprintf("$%d", n)
+}
 
-	return query, nil
+func (f sqlFormatter) SanitiseIdentifier(ident string) string {
+	// TODO: sanitise
+	return ident
 }
 
 func NewSQLFormatter() utils.SQLFormatter {
diff --git a/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter_test.go b/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter_test.go
index 16cd042fc4..1322dc04ea 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter_test.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/postgresql/sql_formatter_test.go
@@ -330,7 +330,7 @@ func TestSQLFormatter(t *testing.T) {
 		tc := tc
 
 		t.Run(tc.testName, func(t *testing.T) {
-			output, err := formatter.FormatRead(logger, tc.selectReq)
+			output, _, err := utils.MakeReadSplitQuery(logger, formatter, tc.selectReq)
 			require.Equal(t, tc.output, output)
 
 			if tc.err != nil {
diff --git a/ydb/library/yql/providers/generic/connector/app/server/postgresql/ya.make b/ydb/library/yql/providers/generic/connector/app/server/postgresql/ya.make
index a08bd57be7..46287aade5 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/postgresql/ya.make
+++ b/ydb/library/yql/providers/generic/connector/app/server/postgresql/ya.make
@@ -3,7 +3,6 @@ GO_LIBRARY()
 SRCS(
     connection_manager.go
     doc.go
-    query_executor.go
     sql_formatter.go
     type_mapper.go
 )
diff --git a/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler.go b/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler.go
index 914a418133..b8f74335df 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler.go
@@ -13,7 +13,6 @@ import (
 type handlerImpl struct {
 	typeMapper        utils.TypeMapper
 	sqlFormatter      utils.SQLFormatter
-	queryBuilder      utils.QueryExecutor
 	connectionManager utils.ConnectionManager
 	logger            log.Logger
 }
@@ -23,6 +22,8 @@ func (h *handlerImpl) DescribeTable(
 	logger log.Logger,
 	request *api_service_protos.TDescribeTableRequest,
 ) (*api_service_protos.TDescribeTableResponse, error) {
+	query, args := utils.MakeDescribeTableQuery(logger, h.sqlFormatter, request)
+
 	conn, err := h.connectionManager.Make(ctx, logger, request.DataSourceInstance)
 	if err != nil {
 		return nil, fmt.Errorf("make connection: %w", err)
@@ -30,7 +31,7 @@ func (h *handlerImpl) DescribeTable(
 
 	defer h.connectionManager.Release(logger, conn)
 
-	rows, err := h.queryBuilder.DescribeTable(ctx, conn, request)
+	rows, err := conn.Query(ctx, query, args...)
 	if err != nil {
 		return nil, fmt.Errorf("query builder error: %w", err)
 	}
@@ -72,7 +73,7 @@ func (h *handlerImpl) doReadSplit(
 	split *api_service_protos.TSplit,
 	sink paging.Sink,
 ) error {
-	query, err := h.sqlFormatter.FormatRead(logger, split.Select)
+	query, args, err := utils.MakeReadSplitQuery(logger, h.sqlFormatter, split.Select)
 	if err != nil {
 		return fmt.Errorf("make read split query: %w", err)
 	}
@@ -84,7 +85,7 @@ func (h *handlerImpl) doReadSplit(
 
 	defer h.connectionManager.Release(logger, conn)
 
-	rows, err := conn.Query(ctx, query)
+	rows, err := conn.Query(ctx, query, args...)
 	if err != nil {
 		return fmt.Errorf("query '%s' error: %w", query, err)
 	}
@@ -136,7 +137,6 @@ func newHandler(
 	return &handlerImpl{
 		logger:            logger,
 		sqlFormatter:      preset.sqlFormatter,
-		queryBuilder:      preset.queryExecutor,
 		connectionManager: preset.connectionManager,
 		typeMapper:        preset.typeMapper,
 	}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler_factory.go b/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler_factory.go
index 33560cebea..a1d5fd1b96 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler_factory.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/rdbms/handler_factory.go
@@ -12,7 +12,6 @@ import (
 
 type handlerPreset struct {
 	sqlFormatter      utils.SQLFormatter
-	queryExecutor     utils.QueryExecutor
 	connectionManager utils.ConnectionManager
 	typeMapper        utils.TypeMapper
 }
@@ -44,13 +43,11 @@ func NewHandlerFactory(qlf utils.QueryLoggerFactory) HandlerFactory {
 	return &handlerFactoryImpl{
 		clickhouse: handlerPreset{
 			sqlFormatter:      clickhouse.NewSQLFormatter(),
-			queryExecutor:     clickhouse.NewQueryExecutor(),
 			connectionManager: clickhouse.NewConnectionManager(connManagerCfg),
 			typeMapper:        clickhouse.NewTypeMapper(),
 		},
 		postgresql: handlerPreset{
 			sqlFormatter:      postgresql.NewSQLFormatter(),
-			queryExecutor:     postgresql.NewQueryExecutor(),
 			connectionManager: postgresql.NewConnectionManager(connManagerCfg),
 			typeMapper:        postgresql.NewTypeMapper(),
 		},
diff --git a/ydb/library/yql/providers/generic/connector/app/server/rdbms/mock.go b/ydb/library/yql/providers/generic/connector/app/server/rdbms/mock.go
index dbe05948dd..73f63b22ad 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/rdbms/mock.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/rdbms/mock.go
@@ -42,7 +42,6 @@ var _ HandlerFactory = (*HandlerFactoryMock)(nil)
 
 type HandlerFactoryMock struct {
 	SQLFormatter      utils.SQLFormatter
-	QueryExecutor     utils.QueryExecutor
 	ConnectionManager utils.ConnectionManager
 	TypeMapper        utils.TypeMapper
 }
@@ -52,7 +51,6 @@ func (m *HandlerFactoryMock) Make(logger log.Logger, dataSourceType api_common.E
 		logger,
 		&handlerPreset{
 			sqlFormatter:      m.SQLFormatter,
-			queryExecutor:     m.QueryExecutor,
 			connectionManager: m.ConnectionManager,
 			typeMapper:        m.TypeMapper,
 		},
diff --git a/ydb/library/yql/providers/generic/connector/app/server/utils/predicate_builder.go b/ydb/library/yql/providers/generic/connector/app/server/utils/predicate_builder.go
index b763a9a265..6ed1aebad8 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/utils/predicate_builder.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/utils/predicate_builder.go
@@ -8,12 +8,7 @@ import (
 	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
 )
 
-type PredicateBuilderFeatures interface {
-	// Support for high level expression (without subexpressions, they are checked separately)
-	SupportsExpression(expression *api_service_protos.TExpression) bool
-}
-
-func FormatValue(value *Ydb.TypedValue) (string, error) {
+func formatValue(value *Ydb.TypedValue) (string, error) {
 	switch v := value.Value.Value.(type) {
 	case *Ydb.Value_BoolValue:
 		return fmt.Sprintf("%t", v.BoolValue), nil
@@ -34,15 +29,15 @@ func FormatValue(value *Ydb.TypedValue) (string, error) {
 	}
 }
 
-func FormatColumn(col string) (string, error) {
+func formatColumn(col string) (string, error) {
 	return col, nil
 }
 
-func FormatNull(n *api_service_protos.TExpression_TNull) (string, error) {
+func formatNull(n *api_service_protos.TExpression_TNull) (string, error) {
 	return "NULL", nil
 }
 
-func FormatArithmeticalExpression(expression *api_service_protos.TExpression_TArithmeticalExpression, features PredicateBuilderFeatures) (string, error) {
+func formatArithmeticalExpression(formatter SQLFormatter, expression *api_service_protos.TExpression_TArithmeticalExpression) (string, error) {
 	var operation string
 
 	switch op := expression.Operation; op {
@@ -68,12 +63,12 @@ func FormatArithmeticalExpression(expression *api_service_protos.TExpression_TAr
 		err   error
 	)
 
-	left, err = FormatExpression(expression.LeftValue, features)
+	left, err = formatExpression(formatter, expression.LeftValue)
 	if err != nil {
 		return "", fmt.Errorf("failed to format left argument: %w", err)
 	}
 
-	right, err = FormatExpression(expression.RightValue, features)
+	right, err = formatExpression(formatter, expression.RightValue)
 	if err != nil {
 		return "", fmt.Errorf("failed to format right argument: %w", err)
 	}
@@ -81,26 +76,26 @@ func FormatArithmeticalExpression(expression *api_service_protos.TExpression_TAr
 	return fmt.Sprintf("(%s%s%s)", left, operation, right), nil
 }
 
-func FormatExpression(expression *api_service_protos.TExpression, features PredicateBuilderFeatures) (string, error) {
-	if !features.SupportsExpression(expression) {
+func formatExpression(formatter SQLFormatter, expression *api_service_protos.TExpression) (string, error) {
+	if !formatter.SupportsPushdownExpression(expression) {
 		return "", ErrUnsupportedExpression
 	}
 
 	switch e := expression.Payload.(type) {
 	case *api_service_protos.TExpression_Column:
-		return FormatColumn(e.Column)
+		return formatColumn(e.Column)
 	case *api_service_protos.TExpression_TypedValue:
-		return FormatValue(e.TypedValue)
+		return formatValue(e.TypedValue)
 	case *api_service_protos.TExpression_ArithmeticalExpression:
-		return FormatArithmeticalExpression(e.ArithmeticalExpression, features)
+		return formatArithmeticalExpression(formatter, e.ArithmeticalExpression)
 	case *api_service_protos.TExpression_Null:
-		return FormatNull(e.Null)
+		return formatNull(e.Null)
 	default:
 		return "", fmt.Errorf("%w, type: %T", ErrUnimplementedExpression, e)
 	}
 }
 
-func FormatComparison(comparison *api_service_protos.TPredicate_TComparison, features PredicateBuilderFeatures) (string, error) {
+func formatComparison(formatter SQLFormatter, comparison *api_service_protos.TPredicate_TComparison) (string, error) {
 	var operation string
 
 	switch op := comparison.Operation; op {
@@ -126,12 +121,12 @@ func FormatComparison(comparison *api_service_protos.TPredicate_TComparison, fea
 		err   error
 	)
 
-	left, err = FormatExpression(comparison.LeftValue, features)
+	left, err = formatExpression(formatter, comparison.LeftValue)
 	if err != nil {
 		return "", fmt.Errorf("failed to format left argument: %w", err)
 	}
 
-	right, err = FormatExpression(comparison.RightValue, features)
+	right, err = formatExpression(formatter, comparison.RightValue)
 	if err != nil {
 		return "", fmt.Errorf("failed to format right argument: %w", err)
 	}
@@ -139,8 +134,8 @@ func FormatComparison(comparison *api_service_protos.TPredicate_TComparison, fea
 	return fmt.Sprintf("(%s%s%s)", left, operation, right), nil
 }
 
-func FormatNegation(negation *api_service_protos.TPredicate_TNegation, features PredicateBuilderFeatures) (string, error) {
-	pred, err := FormatPredicate(negation.Operand, features, false)
+func formatNegation(formatter SQLFormatter, negation *api_service_protos.TPredicate_TNegation) (string, error) {
+	pred, err := formatPredicate(formatter, negation.Operand, false)
 	if err != nil {
 		return "", fmt.Errorf("failed to format NOT statement: %w", err)
 	}
@@ -148,7 +143,7 @@ func FormatNegation(negation *api_service_protos.TPredicate_TNegation, features
 	return fmt.Sprintf("(NOT %s)", pred), nil
 }
 
-func FormatConjunction(conjunction *api_service_protos.TPredicate_TConjunction, features PredicateBuilderFeatures, topLevel bool) (string, error) {
+func formatConjunction(formatter SQLFormatter, conjunction *api_service_protos.TPredicate_TConjunction, topLevel bool) (string, error) {
 	var (
 		sb        strings.Builder
 		succeeded int32 = 0
@@ -158,7 +153,7 @@ func FormatConjunction(conjunction *api_service_protos.TPredicate_TConjunction,
 	)
 
 	for _, predicate := range conjunction.Operands {
-		statement, err = FormatPredicate(predicate, features, false)
+		statement, err = formatPredicate(formatter, predicate, false)
 		if err != nil {
 			if !topLevel {
 				return "", fmt.Errorf("failed to format AND statement: %w", err)
@@ -193,7 +188,7 @@ func FormatConjunction(conjunction *api_service_protos.TPredicate_TConjunction,
 	return sb.String(), nil
 }
 
-func FormatDisjunction(disjunction *api_service_protos.TPredicate_TDisjunction, features PredicateBuilderFeatures) (string, error) {
+func formatDisjunction(formatter SQLFormatter, disjunction *api_service_protos.TPredicate_TDisjunction) (string, error) {
 	var (
 		sb        strings.Builder
 		cnt       int32 = 0
@@ -203,7 +198,7 @@ func FormatDisjunction(disjunction *api_service_protos.TPredicate_TDisjunction,
 	)
 
 	for _, predicate := range disjunction.Operands {
-		statement, err = FormatPredicate(predicate, features, false)
+		statement, err = formatPredicate(formatter, predicate, false)
 		if err != nil {
 			return "", fmt.Errorf("failed to format OR statement: %w", err)
 		} else {
@@ -236,8 +231,8 @@ func FormatDisjunction(disjunction *api_service_protos.TPredicate_TDisjunction,
 	return sb.String(), nil
 }
 
-func FormatIsNull(isNull *api_service_protos.TPredicate_TIsNull, features PredicateBuilderFeatures) (string, error) {
-	statement, err := FormatExpression(isNull.Value, features)
+func formatIsNull(formatter SQLFormatter, isNull *api_service_protos.TPredicate_TIsNull) (string, error) {
+	statement, err := formatExpression(formatter, isNull.Value)
 	if err != nil {
 		return "", fmt.Errorf("failed to format IS NULL statement: %w", err)
 	}
@@ -245,8 +240,8 @@ func FormatIsNull(isNull *api_service_protos.TPredicate_TIsNull, features Predic
 	return fmt.Sprintf("(%s IS NULL)", statement), nil
 }
 
-func FormatIsNotNull(isNotNull *api_service_protos.TPredicate_TIsNotNull, features PredicateBuilderFeatures) (string, error) {
-	statement, err := FormatExpression(isNotNull.Value, features)
+func formatIsNotNull(formatter SQLFormatter, isNotNull *api_service_protos.TPredicate_TIsNotNull) (string, error) {
+	statement, err := formatExpression(formatter, isNotNull.Value)
 	if err != nil {
 		return "", fmt.Errorf("failed to format IS NOT NULL statement: %w", err)
 	}
@@ -254,33 +249,33 @@ func FormatIsNotNull(isNotNull *api_service_protos.TPredicate_TIsNotNull, featur
 	return fmt.Sprintf("(%s IS NOT NULL)", statement), nil
 }
 
-func FormatPredicate(predicate *api_service_protos.TPredicate, features PredicateBuilderFeatures, topLevel bool) (string, error) {
+func formatPredicate(formatter SQLFormatter, predicate *api_service_protos.TPredicate, topLevel bool) (string, error) {
 	switch p := predicate.Payload.(type) {
 	case *api_service_protos.TPredicate_Negation:
-		return FormatNegation(p.Negation, features)
+		return formatNegation(formatter, p.Negation)
 	case *api_service_protos.TPredicate_Conjunction:
-		return FormatConjunction(p.Conjunction, features, topLevel)
+		return formatConjunction(formatter, p.Conjunction, topLevel)
 	case *api_service_protos.TPredicate_Disjunction:
-		return FormatDisjunction(p.Disjunction, features)
+		return formatDisjunction(formatter, p.Disjunction)
 	case *api_service_protos.TPredicate_IsNull:
-		return FormatIsNull(p.IsNull, features)
+		return formatIsNull(formatter, p.IsNull)
 	case *api_service_protos.TPredicate_IsNotNull:
-		return FormatIsNotNull(p.IsNotNull, features)
+		return formatIsNotNull(formatter, p.IsNotNull)
 	case *api_service_protos.TPredicate_Comparison:
-		return FormatComparison(p.Comparison, features)
+		return formatComparison(formatter, p.Comparison)
 	case *api_service_protos.TPredicate_BoolExpression:
-		return FormatExpression(p.BoolExpression.Value, features)
+		return formatExpression(formatter, p.BoolExpression.Value)
 	default:
 		return "", fmt.Errorf("%w, type: %T", ErrUnimplementedPredicateType, p)
 	}
 }
 
-func FormatWhereClause(where *api_service_protos.TSelect_TWhere, features PredicateBuilderFeatures) (string, error) {
+func formatWhereClause(formatter SQLFormatter, where *api_service_protos.TSelect_TWhere) (string, error) {
 	if where.FilterTyped == nil {
 		return "", ErrUnimplemented
 	}
 
-	formatted, err := FormatPredicate(where.FilterTyped, features, true)
+	formatted, err := formatPredicate(formatter, where.FilterTyped, true)
 	if err != nil {
 		return "", err
 	}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/utils/query_builder.go b/ydb/library/yql/providers/generic/connector/app/server/utils/query_builder.go
new file mode 100644
index 0000000000..5fad123291
--- /dev/null
+++ b/ydb/library/yql/providers/generic/connector/app/server/utils/query_builder.go
@@ -0,0 +1,40 @@
+package utils
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/ydb-platform/ydb/library/go/core/log"
+	api_service_protos "github.com/ydb-platform/ydb/ydb/library/yql/providers/generic/connector/libgo/service/protos"
+)
+
+func MakeDescribeTableQuery(logger log.Logger, formatter SQLFormatter, request *api_service_protos.TDescribeTableRequest) (string, []any) {
+	query, args := formatter.GetDescribeTableQuery(request)
+
+	return query, args
+}
+
+func MakeReadSplitQuery(logger log.Logger, formatter SQLFormatter, request *api_service_protos.TSelect) (string, []any, error) {
+	var sb strings.Builder
+
+	selectPart, err := formatSelectColumns(formatter, request.What, request.GetFrom().GetTable(), true)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to format select statement: %w", err)
+	}
+
+	sb.WriteString(selectPart)
+
+	if request.Where != nil {
+		clause, err := formatWhereClause(formatter, request.Where)
+		if err != nil {
+			logger.Error("Failed to format WHERE clause", log.Error(err), log.String("where", request.Where.String()))
+		} else {
+			sb.WriteString(" ")
+			sb.WriteString(clause)
+		}
+	}
+
+	query := sb.String()
+
+	return query, nil, nil
+}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/utils/select_helpers.go b/ydb/library/yql/providers/generic/connector/app/server/utils/select_helpers.go
index ebf5e7d0c1..352da3fab0 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/utils/select_helpers.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/utils/select_helpers.go
@@ -23,7 +23,7 @@ func SelectWhatToYDBTypes(selectWhat *api_service_protos.TSelect_TWhat) ([]*Ydb.
 	return ydbTypes, nil
 }
 
-func SelectWhatToYDBColumns(selectWhat *api_service_protos.TSelect_TWhat) ([]*Ydb.Column, error) {
+func selectWhatToYDBColumns(selectWhat *api_service_protos.TSelect_TWhat) ([]*Ydb.Column, error) {
 	var columns []*Ydb.Column
 
 	for i, item := range selectWhat.Items {
@@ -38,7 +38,7 @@ func SelectWhatToYDBColumns(selectWhat *api_service_protos.TSelect_TWhat) ([]*Yd
 	return columns, nil
 }
 
-func FormatSelectColumns(selectWhat *api_service_protos.TSelect_TWhat, tableName string, fakeZeroOnEmptyColumnsSet bool) (string, error) {
+func formatSelectColumns(formatter SQLFormatter, selectWhat *api_service_protos.TSelect_TWhat, tableName string, fakeZeroOnEmptyColumnsSet bool) (string, error) {
 	// SELECT $columns FROM $from
 	if tableName == "" {
 		return "", ErrEmptyTableName
@@ -48,7 +48,7 @@ func FormatSelectColumns(selectWhat *api_service_protos.TSelect_TWhat, tableName
 
 	sb.WriteString("SELECT ")
 
-	columns, err := SelectWhatToYDBColumns(selectWhat)
+	columns, err := selectWhatToYDBColumns(selectWhat)
 	if err != nil {
 		return "", fmt.Errorf("convert Select.What.Items to Ydb.Columns: %w", err)
 	}
diff --git a/ydb/library/yql/providers/generic/connector/app/server/utils/sql.go b/ydb/library/yql/providers/generic/connector/app/server/utils/sql.go
index 51d7c2da76..d4b015da3a 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/utils/sql.go
+++ b/ydb/library/yql/providers/generic/connector/app/server/utils/sql.go
@@ -30,10 +30,11 @@ type ConnectionManagerBase struct {
 	QueryLoggerFactory QueryLoggerFactory
 }
 
-type QueryExecutor interface {
-	DescribeTable(ctx context.Context, conn Connection, request *api_service_protos.TDescribeTableRequest) (Rows, error)
-}
-
 type SQLFormatter interface {
-	FormatRead(logger log.Logger, selectReq *api_service_protos.TSelect) (string, error)
+	GetDescribeTableQuery(request *api_service_protos.TDescribeTableRequest) (string, []any)
+	GetPlaceholder(n int) string
+	SanitiseIdentifier(ident string) string
+
+	// Support for high level expression (without subexpressions, they are checked separately)
+	SupportsPushdownExpression(expression *api_service_protos.TExpression) bool
 }
diff --git a/ydb/library/yql/providers/generic/connector/app/server/utils/ya.make b/ydb/library/yql/providers/generic/connector/app/server/utils/ya.make
index a2c30d45ee..28eafde38a 100644
--- a/ydb/library/yql/providers/generic/connector/app/server/utils/ya.make
+++ b/ydb/library/yql/providers/generic/connector/app/server/utils/ya.make
@@ -10,6 +10,7 @@ SRCS(
     logger.go
     predicate_builder.go
     protobuf.go
+    query_builder.go
     select_helpers.go
     sql.go
     sql_mock.go