Logging: virtual table schema

app/server/datasource/rdbms/clickhouse/connection_http.go

@@ -26,7 +26,7 @@ type rows struct {
 	*sql.Rows
 }
 
-func (r *rows) MakeTransformer(ydbTypes []*Ydb.Type, cc conversion.Collection) (paging.RowTransformer[any], error) {
+func (r *rows) MakeTransformer(ydbColumns []*Ydb.Column, cc conversion.Collection) (paging.RowTransformer[any], error) {
 	columns, err := r.ColumnTypes()
 	if err != nil {
 		return nil, fmt.Errorf("column types: %w", err)
@@ -37,7 +37,7 @@ func (r *rows) MakeTransformer(ydbTypes []*Ydb.Type, cc conversion.Collection) (
 		typeNames = append(typeNames, column.DatabaseTypeName())
 	}
 
-	transformer, err := transformerFromSQLTypes(typeNames, ydbTypes, cc)
+	transformer, err := transformerFromSQLTypes(typeNames, common.YDBColumnsToYDBTypes(ydbColumns), cc)
 	if err != nil {
 		return nil, fmt.Errorf("transformer from sql types: %w", err)
 	}

app/server/datasource/rdbms/clickhouse/connection_native.go

@@ -29,15 +29,15 @@ func (rowsNative) NextResultSet() bool {
 	return false
 }
 
-func (r *rowsNative) MakeTransformer(ydbTypes []*Ydb.Type, cc conversion.Collection) (paging.RowTransformer[any], error) {
+func (r *rowsNative) MakeTransformer(ydbColumns []*Ydb.Column, cc conversion.Collection) (paging.RowTransformer[any], error) {
 	columns := r.ColumnTypes()
 
 	typeNames := make([]string, 0, len(columns))
 	for _, column := range columns {
 		typeNames = append(typeNames, column.DatabaseTypeName())
 	}
 
-	transformer, err := transformerFromSQLTypes(typeNames, ydbTypes, cc)
+	transformer, err := transformerFromSQLTypes(typeNames, common.YDBColumnsToYDBTypes(ydbColumns), cc)
 	if err != nil {
 		return nil, fmt.Errorf("transformer from sql types: %w", err)
 	}

app/server/datasource/rdbms/clickhouse/sql_formatter_test.go

@@ -477,7 +477,7 @@ func TestMakeSelectQuery(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, tc.outputQuery, readSplitsQuery.QueryText)
 			require.Equal(t, tc.outputArgs, readSplitsQuery.QueryArgs.Values())
-			require.Equal(t, tc.outputYdbTypes, readSplitsQuery.YdbTypes)
+			require.Equal(t, tc.outputYdbTypes, common.YDBColumnsToYDBTypes(readSplitsQuery.YdbColumns))
 		})
 	}
 }

app/server/datasource/rdbms/datasource.go

@@ -246,7 +246,7 @@ func (ds *dataSourceImpl) doReadSplitSingleConn(
 
 	defer common.LogCloserError(logger, rows, "close rows")
 
-	transformer, err := rows.MakeTransformer(query.YdbTypes, ds.converterCollection)
+	transformer, err := rows.MakeTransformer(query.YdbColumns, ds.converterCollection)
 	if err != nil {
 		return 0, fmt.Errorf("make transformer: %w", err)
 	}

app/server/datasource/rdbms/datasource_factory.go

@@ -206,8 +206,8 @@ func NewDataSourceFactory(
 	dsf.logging = Preset{
 		SQLFormatter:      logging.NewSQLFormatter(ydb.NewSQLFormatter(cfg.Logging.Ydb.Mode, cfg.Logging.Ydb.Pushdown)),
 		ConnectionManager: logging.NewConnectionManager(cfg.Logging, connManagerBase, dsf.loggingResolver),
-		TypeMapper:        ydbTypeMapper,
-		SchemaProvider:    ydb.NewSchemaProvider(ydbTypeMapper),
+		TypeMapper:        nil,
+		SchemaProvider:    logging.NewSchemaProvider(),
 		SplitProvider:     logging.NewSplitProvider(dsf.loggingResolver, ydb.NewSplitProvider(cfg.Logging.Ydb.Splitting)),
 		RetrierSet: &retry.RetrierSet{
 			MakeConnection: retry.NewRetrierFromConfig(cfg.Ydb.ExponentialBackoff, retry.ErrorCheckerMakeConnectionCommon),

app/server/datasource/rdbms/datasource_test.go

@@ -93,7 +93,16 @@ func TestReadSplit(t *testing.T) {
 		}
 
 		rows.On("MakeTransformer",
-			[]*Ydb.Type{common.MakePrimitiveType(Ydb.Type_INT32), common.MakePrimitiveType(Ydb.Type_UTF8)},
+			[]*Ydb.Column{
+				{
+					Name: "col1",
+					Type: common.MakePrimitiveType(Ydb.Type_INT32),
+				},
+				{
+					Name: "col2",
+					Type: common.MakePrimitiveType(Ydb.Type_UTF8),
+				},
+			},
 		).Return(transformer, nil).Once()
 		rows.On("Next").Return(true).Times(2)
 		rows.On("Next").Return(false).Once()
@@ -158,9 +167,15 @@ func TestReadSplit(t *testing.T) {
 		scanErr := fmt.Errorf("scan failed")
 
 		rows.On("MakeTransformer",
-			[]*Ydb.Type{
-				common.MakePrimitiveType(Ydb.Type_INT32),
-				common.MakePrimitiveType(Ydb.Type_UTF8),
+			[]*Ydb.Column{
+				{
+					Name: "col1",
+					Type: common.MakePrimitiveType(Ydb.Type_INT32),
+				},
+				{
+					Name: "col2",
+					Type: common.MakePrimitiveType(Ydb.Type_UTF8),
+				},
 			},
 		).Return(transformer, nil).Once()
 		rows.On("Next").Return(true).Times(2)

app/server/datasource/rdbms/logging/connection.go

@@ -0,0 +1,37 @@
+package logging
+
+import (
+	"fmt"
+
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+
+	"github.com/ydb-platform/fq-connector-go/app/server/conversion"
+	rdbms_utils "github.com/ydb-platform/fq-connector-go/app/server/datasource/rdbms/utils"
+	"github.com/ydb-platform/fq-connector-go/app/server/paging"
+)
+
+var _ rdbms_utils.Rows = (*rowsImpl)(nil)
+
+type rowsImpl struct {
+	rdbms_utils.Rows
+}
+
+func (rowsImpl) MakeTransformer(ydbColumns []*Ydb.Column, cc conversion.Collection) (paging.RowTransformer[any], error) {
+	return makeRowTransformer(ydbColumns, cc)
+}
+
+var _ rdbms_utils.Connection = (*connectionImpl)(nil)
+
+type connectionImpl struct {
+	rdbms_utils.Connection
+}
+
+func (c *connectionImpl) Query(params *rdbms_utils.QueryParams) (rdbms_utils.Rows, error) {
+	ydbRows, err := c.Connection.Query(params)
+	if err != nil {
+		return nil, fmt.Errorf("ydb connection query: %w", err)
+	}
+
+	// Wrap YDB row iterator with new implementation that uses custom transformer
+	return &rowsImpl{Rows: ydbRows}, nil
+}

app/server/datasource/rdbms/logging/connection_manager.go

@@ -16,7 +16,6 @@
 )
 
 type connectionManager struct {
-	rdbms_utils.Connection
 	ydbConnectionManager rdbms_utils.ConnectionManager
 	resolver             Resolver
 }
@@ -121,12 +120,18 @@
 		UseTls:      true,
 	}
 
-	cs, err := cm.makeConnectionFromDataSourceInstance(params.Ctx, params.Logger, dsi, src.TableName)
+	ydbConns, err := cm.makeConnectionFromDataSourceInstance(params.Ctx, params.Logger, dsi, src.TableName)
 	if err != nil {
 		return nil, fmt.Errorf("make connection from data source instance: %w", err)
 	}
 
-	return cs, nil
+	// Wrap YDB connections with Logging decorator to override data types
+	loggingCons := make([]rdbms_utils.Connection, 0, len(ydbConns))
+	for _, ydbConn := range ydbConns {
+		loggingCons = append(loggingCons, &connectionImpl{Connection: ydbConn})
+	}
+
+	return loggingCons, nil
 }
 
 func (cm *connectionManager) makeConnectionFromDataSourceInstance(

app/server/datasource/rdbms/logging/consts.go

@@ -0,0 +1,15 @@
+package logging
+
+const (
+	levelColumnName     = "level"
+	timestampColumnName = "timestamp"
+	messageColumnName   = "message"
+	metaColumnName      = "meta"
+
+	levelTraceValue = "TRACE"
+	levelDebugValue = "DEBUG"
+	levelInfoValue  = "INFO"
+	levelWarnValue  = "WARN"
+	levelErrorValue = "ERROR"
+	levelFatalValue = "FATAL"
+)

app/server/datasource/rdbms/logging/schema_provider.go

@@ -0,0 +1,40 @@
+package logging
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+
+	api_service_protos "github.com/ydb-platform/fq-connector-go/api/service/protos"
+	rdbms_utils "github.com/ydb-platform/fq-connector-go/app/server/datasource/rdbms/utils"
+	"github.com/ydb-platform/fq-connector-go/common"
+)
+
+var loggingVirtualSchema = &api_service_protos.TSchema{
+	Columns: []*Ydb.Column{
+		{Name: levelColumnName, Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_UTF8))},
+		{Name: timestampColumnName, Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_TIMESTAMP))},
+		{Name: messageColumnName, Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_UTF8))},
+		{Name: metaColumnName, Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_JSON))},
+	},
+}
+
+var _ rdbms_utils.SchemaProvider = (*schemaProviderImpl)(nil)
+
+type schemaProviderImpl struct {
+}
+
+func (schemaProviderImpl) GetSchema(
+	_ context.Context,
+	_ *zap.Logger,
+	_ rdbms_utils.Connection,
+	_ *api_service_protos.TDescribeTableRequest,
+) (*api_service_protos.TSchema, error) {
+	return loggingVirtualSchema, nil
+}
+
+func NewSchemaProvider() rdbms_utils.SchemaProvider {
+	return &schemaProviderImpl{}
+}

app/server/datasource/rdbms/logging/sql_formatter.go

@@ -4,10 +4,14 @@
 	"fmt"
 
 	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
 
 	api_service_protos "github.com/ydb-platform/fq-connector-go/api/service/protos"
 	rdbms_utils "github.com/ydb-platform/fq-connector-go/app/server/datasource/rdbms/utils"
 	"github.com/ydb-platform/fq-connector-go/app/server/datasource/rdbms/ydb"
+	"github.com/ydb-platform/fq-connector-go/common"
 )
 
 var _ rdbms_utils.SQLFormatter = (*sqlFormatter)(nil)
@@ -29,6 +33,95 @@
 	return s.RenderSelectQueryTextForColumnShard(parts, dst.GetYdb().TabletIds)
 }
 
+func (sqlFormatter) TransformSelectWhat(src *api_service_protos.TSelect_TWhat) *api_service_protos.TSelect_TWhat {
+	dst := &api_service_protos.TSelect_TWhat{}
+
+	for _, item := range src.GetItems() {
+		name := item.GetColumn().Name
+		switch name {
+		case metaColumnName:
+			dst.Items = append(dst.Items, &api_service_protos.TSelect_TWhat_TItem{
+				Payload: &api_service_protos.TSelect_TWhat_TItem_Column{
+					Column: &Ydb.Column{
+						Name: "json_payload",
+						Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_JSON)),
+					},
+				},
+			})
+		case levelColumnName:
+			dst.Items = append(dst.Items, &api_service_protos.TSelect_TWhat_TItem{
+				Payload: &api_service_protos.TSelect_TWhat_TItem_Column{
+					Column: &Ydb.Column{
+						Name: "level",
+						Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_INT32)),
+					},
+				},
+			})
+		default:
+			dst.Items = append(dst.Items, item)
+		}
+	}
+
+	return dst
+}
+
+func (sqlFormatter) TransformPredicateComparison(
+	src *api_service_protos.TPredicate_TComparison,
+) (*api_service_protos.TPredicate_TComparison, error) {
+	dst := proto.Clone(src).(*api_service_protos.TPredicate_TComparison)
+
+	// For the comparison related to `level` field
+	if src.LeftValue.GetColumn() == levelColumnName && src.RightValue.GetTypedValue() != nil {
+		if src.Operation != api_service_protos.TPredicate_TComparison_EQ {
+			return nil, fmt.Errorf("unsupported operation %v for `level` column comparison", src.Operation)
+		}
+
+		// Extract filter value of a string type
+		var levelValue string
+
+		switch src.RightValue.GetTypedValue().GetType().GetTypeId() {
+		case Ydb.Type_UTF8:
+			levelValue = src.RightValue.GetTypedValue().GetValue().GetTextValue()
+		case Ydb.Type_STRING:
+			levelValue = string(src.RightValue.GetTypedValue().GetValue().GetBytesValue())
+		default:
+			return nil, fmt.Errorf(
+				"unsupported typed value of type %v for `level` column comparison",
+				src.RightValue.GetTypedValue().GetType(),
+			)
+		}
+
+		// Replace it with number representation
+		switch levelValue {
+		case levelTraceValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(1)
+		case levelDebugValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(2)
+		case levelInfoValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(3)
+		case levelWarnValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(4)
+		case levelErrorValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(5)
+		case levelFatalValue:
+			dst.RightValue.Payload = makeTypedValueForLevel(6)
+		default:
+			return nil, fmt.Errorf("unsupported `level` value %s", levelValue)
+		}
+	}
+
+	return dst, nil
+}
+
+func makeTypedValueForLevel(level int32) *api_service_protos.TExpression_TypedValue {
+	return &api_service_protos.TExpression_TypedValue{
+		TypedValue: common.MakeTypedValue(
+			common.MakePrimitiveType(Ydb.Type_INT32),
+			level,
+		),
+	}
+}
+
 func NewSQLFormatter(ydbSQLFormatter ydb.SQLFormatter) rdbms_utils.SQLFormatter {
 	return &sqlFormatter{
 		SQLFormatter: ydbSQLFormatter,