Merge branch 'main' into promoted-paths

.gitignore

@@ -49,6 +49,7 @@ ee/query-service/tests/test-deploy/data/
 # local data
 *.backup
 *.db
+**/db
 /deploy/docker/clickhouse-setup/data/
 /deploy/docker-swarm/clickhouse-setup/data/
 bin/

Makefile

@@ -72,6 +72,12 @@ devenv-up: devenv-clickhouse devenv-signoz-otel-collector ## Start both clickhou
 	@echo "   - ClickHouse: http://localhost:8123"
 	@echo "   - Signoz OTel Collector: grpc://localhost:4317, http://localhost:4318"
 
+.PHONY: devenv-clickhouse-clean
+devenv-clickhouse-clean: ## Clean all ClickHouse data from filesystem
+	@echo "Removing ClickHouse data..."
+	@rm -rf .devenv/docker/clickhouse/fs/tmp/*
+	@echo "ClickHouse data cleaned!"
+
 ##############################################################
 # go commands
 ##############################################################

pkg/errors/errors.go

@@ -209,6 +209,11 @@ func NewUnexpectedf(code Code, format string, args ...any) *base {
 	return Newf(TypeInvalidInput, code, format, args...)
 }
 
+// NewMethodNotAllowedf is a wrapper around Newf with TypeMethodNotAllowed.
+func NewMethodNotAllowedf(code Code, format string, args ...any) *base {
+	return Newf(TypeMethodNotAllowed, code, format, args...)
+}
+
 // WrapTimeoutf is a wrapper around Wrapf with TypeTimeout.
 func WrapTimeoutf(cause error, code Code, format string, args ...any) *base {
 	return Wrapf(cause, TypeTimeout, code, format, args...)

pkg/modules/promote/implpromote/handler.go

@@ -0,0 +1,60 @@
+package implpromote
+
+import (
+	"net/http"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/http/binding"
+	"github.com/SigNoz/signoz/pkg/http/render"
+	"github.com/SigNoz/signoz/pkg/modules/promote"
+	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+)
+
+type handler struct {
+	module promote.Module
+}
+
+func NewHandler(module promote.Module) promote.Handler {
+	return &handler{module: module}
+}
+
+func (h *handler) HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Request) {
+	// TODO(Nitya): Use in multi tenant setup
+	_, err := authtypes.ClaimsFromContext(r.Context())
+	if err != nil {
+		render.Error(w, errors.NewInternalf(errors.CodeInternal, "failed to get org id from context"))
+		return
+	}
+
+	var req []promotetypes.PromotePath
+	if err := binding.JSON.BindBody(r.Body, &req); err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	err = h.module.PromoteAndIndexPaths(r.Context(), req...)
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	render.Success(w, http.StatusCreated, nil)
+}
+
+func (h *handler) ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request) {
+	// TODO(Nitya): Use in multi tenant setup
+	_, err := authtypes.ClaimsFromContext(r.Context())
+	if err != nil {
+		render.Error(w, errors.NewInternalf(errors.CodeInternal, "failed to get org id from context"))
+		return
+	}
+
+	paths, err := h.module.ListPromotedAndIndexedPaths(r.Context())
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	render.Success(w, http.StatusOK, paths)
+}

pkg/modules/promote/implpromote/module.go

@@ -0,0 +1,201 @@
+package implpromote
+
+import (
+	"context"
+	"maps"
+	"slices"
+	"strings"
+
+	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/modules/promote"
+	"github.com/SigNoz/signoz/pkg/telemetrylogs"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+var (
+	CodeFailedToCreateIndex        = errors.MustNewCode("failed_to_create_index_promoted_paths")
+	CodeFailedToQueryPromotedPaths = errors.MustNewCode("failed_to_query_promoted_paths")
+)
+
+type module struct {
+	metadataStore  telemetrytypes.MetadataStore
+	telemetryStore telemetrystore.TelemetryStore
+}
+
+func NewModule(metadataStore telemetrytypes.MetadataStore, telemetrystore telemetrystore.TelemetryStore) promote.Module {
+	return &module{metadataStore: metadataStore, telemetryStore: telemetrystore}
+}
+
+func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetypes.PromotePath, error) {
+	logsIndexes, err := m.metadataStore.ListLogsJSONIndexes(ctx)
+	if err != nil {
+		return nil, err
+	}
+	// Flatten the map values (which are slices) into a single slice
+	indexes := slices.Concat(slices.Collect(maps.Values(logsIndexes))...)
+
+	aggr := map[string][]promotetypes.WrappedIndex{}
+	for _, index := range indexes {
+		path, columnType, err := schemamigrator.UnfoldJSONSubColumnIndexExpr(index.Expression)
+		if err != nil {
+			return nil, err
+		}
+
+		// clean backticks from the path
+		path = strings.ReplaceAll(path, "`", "")
+
+		aggr[path] = append(aggr[path], promotetypes.WrappedIndex{
+			ColumnType:  columnType,
+			Type:        index.Type,
+			Granularity: index.Granularity,
+		})
+	}
+	promotedPaths, err := m.listPromotedPaths(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	response := []promotetypes.PromotePath{}
+	for _, path := range promotedPaths {
+		fullPath := telemetrylogs.BodyPromotedColumnPrefix + path
+		path = telemetrylogs.BodyJSONStringSearchPrefix + path
+		item := promotetypes.PromotePath{
+			Path:    path,
+			Promote: true,
+		}
+		indexes, ok := aggr[fullPath]
+		if ok {
+			item.Indexes = indexes
+			delete(aggr, fullPath)
+		}
+		response = append(response, item)
+	}
+
+	// add the paths that are not promoted but have indexes
+	for path, indexes := range aggr {
+		path := strings.TrimPrefix(path, telemetrylogs.BodyJSONColumnPrefix)
+		path = telemetrylogs.BodyJSONStringSearchPrefix + path
+		response = append(response, promotetypes.PromotePath{
+			Path:    path,
+			Indexes: indexes,
+		})
+	}
+	return response, nil
+}
+
+func (m *module) listPromotedPaths(ctx context.Context) ([]string, error) {
+	paths, err := m.metadataStore.ListPromotedPaths(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return slices.Collect(maps.Keys(paths)), nil
+}
+
+// PromotePaths inserts provided JSON paths into the promoted paths table for logs queries.
+func (m *module) PromotePaths(ctx context.Context, paths []string) error {
+	if len(paths) == 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "paths cannot be empty")
+	}
+
+	return m.metadataStore.PromotePaths(ctx, paths...)
+}
+
+// createIndexes creates string ngram + token filter indexes on JSON path subcolumns for LIKE queries.
+func (m *module) createIndexes(ctx context.Context, indexes []schemamigrator.Index) error {
+	if len(indexes) == 0 {
+		return nil
+	}
+
+	for _, index := range indexes {
+		alterStmt := schemamigrator.AlterTableAddIndex{
+			Database: telemetrylogs.DBName,
+			Table:    telemetrylogs.LogsV2LocalTableName,
+			Index:    index,
+		}
+		op := alterStmt.OnCluster(m.telemetryStore.Cluster())
+		if err := m.telemetryStore.ClickhouseDB().Exec(ctx, op.ToSQL()); err != nil {
+			return errors.WrapInternalf(err, CodeFailedToCreateIndex, "failed to create index")
+		}
+	}
+
+	return nil
+}
+
+// PromoteAndIndexPaths handles promoting paths and creating indexes in one call.
+func (m *module) PromoteAndIndexPaths(
+	ctx context.Context,
+	paths ...promotetypes.PromotePath,
+) error {
+	if len(paths) == 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "paths cannot be empty")
+	}
+
+	pathsStr := []string{}
+	// validate the paths
+	for _, path := range paths {
+		if err := path.Validate(); err != nil {
+			return err
+		}
+		pathsStr = append(pathsStr, path.Path)
+	}
+
+	existingPromotedPaths, err := m.metadataStore.ListPromotedPaths(ctx, pathsStr...)
+	if err != nil {
+		return err
+	}
+
+	var toInsert []string
+	indexes := []schemamigrator.Index{}
+	for _, it := range paths {
+		if it.Promote {
+			if _, promoted := existingPromotedPaths[it.Path]; !promoted {
+				toInsert = append(toInsert, it.Path)
+			}
+		}
+		if len(it.Indexes) > 0 {
+			parentColumn := telemetrylogs.LogsV2BodyJSONColumn
+			// if the path is already promoted or is being promoted, add it to the promoted column
+			if _, promoted := existingPromotedPaths[it.Path]; promoted || it.Promote {
+				parentColumn = telemetrylogs.LogsV2BodyPromotedColumn
+			}
+
+			for _, index := range it.Indexes {
+				var typeIndex schemamigrator.IndexType
+				switch {
+				case strings.HasPrefix(index.Type, string(schemamigrator.IndexTypeNGramBF)):
+					typeIndex = schemamigrator.IndexTypeNGramBF
+				case strings.HasPrefix(index.Type, string(schemamigrator.IndexTypeTokenBF)):
+					typeIndex = schemamigrator.IndexTypeTokenBF
+				case strings.HasPrefix(index.Type, string(schemamigrator.IndexTypeMinMax)):
+					typeIndex = schemamigrator.IndexTypeMinMax
+				default:
+					return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid index type: %s", index.Type)
+				}
+				indexes = append(indexes, schemamigrator.Index{
+					Name:        schemamigrator.JSONSubColumnIndexName(parentColumn, it.Path, index.JSONDataType.StringValue(), typeIndex),
+					Expression:  schemamigrator.JSONSubColumnIndexExpr(parentColumn, it.Path, index.JSONDataType.StringValue()),
+					Type:        index.Type,
+					Granularity: index.Granularity,
+				})
+			}
+		}
+	}
+
+	if len(toInsert) > 0 {
+		err := m.PromotePaths(ctx, toInsert)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(indexes) > 0 {
+		if err := m.createIndexes(ctx, indexes); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

pkg/modules/promote/promote.go

@@ -0,0 +1,18 @@
+package promote
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+)
+
+type Module interface {
+	ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetypes.PromotePath, error)
+	PromoteAndIndexPaths(ctx context.Context, paths ...promotetypes.PromotePath) error
+}
+
+type Handler interface {
+	HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Request)
+	ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request)
+}

pkg/query-service/app/clickhouseReader/reader.go

@@ -43,6 +43,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/query-service/app/traces/tracedetail"
 	"github.com/SigNoz/signoz/pkg/query-service/common"
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
+
 	chErrors "github.com/SigNoz/signoz/pkg/query-service/errors"
 	"github.com/SigNoz/signoz/pkg/query-service/metrics"
 	"github.com/SigNoz/signoz/pkg/query-service/model"

pkg/query-service/app/http_handler.go

@@ -555,6 +555,7 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
 	router.HandleFunc("/api/v1/settings/ttl", am.ViewAccess(aH.getTTL)).Methods(http.MethodGet)
 	router.HandleFunc("/api/v2/settings/ttl", am.AdminAccess(aH.setCustomRetentionTTL)).Methods(http.MethodPost)
 	router.HandleFunc("/api/v2/settings/ttl", am.ViewAccess(aH.getCustomRetentionTTL)).Methods(http.MethodGet)
+
 	router.HandleFunc("/api/v1/settings/apdex", am.AdminAccess(aH.Signoz.Handlers.Apdex.Set)).Methods(http.MethodPost)
 	router.HandleFunc("/api/v1/settings/apdex", am.ViewAccess(aH.Signoz.Handlers.Apdex.Get)).Methods(http.MethodGet)
 
@@ -3991,6 +3992,10 @@ func (aH *APIHandler) RegisterLogsRoutes(router *mux.Router, am *middleware.Auth
 	subRouter.HandleFunc("/pipelines/preview", am.ViewAccess(aH.PreviewLogsPipelinesHandler)).Methods(http.MethodPost)
 	subRouter.HandleFunc("/pipelines/{version}", am.ViewAccess(aH.ListLogsPipelinesHandler)).Methods(http.MethodGet)
 	subRouter.HandleFunc("/pipelines", am.EditAccess(aH.CreateLogsPipeline)).Methods(http.MethodPost)
+
+	// Promote and index JSON paths used in logs
+	subRouter.HandleFunc("/promote_paths", am.AdminAccess(aH.Signoz.Handlers.Promote.ListPromotedAndIndexedPaths)).Methods(http.MethodGet)
+	subRouter.HandleFunc("/promote_paths", am.AdminAccess(aH.Signoz.Handlers.Promote.HandlePromoteAndIndexPaths)).Methods(http.MethodPost)
 }
 
 func (aH *APIHandler) logFields(w http.ResponseWriter, r *http.Request) {

pkg/query-service/app/logs/v4/query_builder_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
+	"github.com/stretchr/testify/assert"
 )
 
 func Test_getClickhouseKey(t *testing.T) {
@@ -1210,9 +1211,8 @@ func TestPrepareLogsQuery(t *testing.T) {
 				t.Errorf("PrepareLogsQuery() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
-			if got != tt.want {
-				t.Errorf("PrepareLogsQuery() = %v, want %v", got, tt.want)
-			}
+
+			assert.Equal(t, tt.want, got)
 		})
 	}
 }

pkg/signoz/handler.go

@@ -9,6 +9,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/dashboard/impldashboard"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer/implmetricsexplorer"
+	"github.com/SigNoz/signoz/pkg/modules/promote"
+	"github.com/SigNoz/signoz/pkg/modules/promote/implpromote"
 	"github.com/SigNoz/signoz/pkg/modules/quickfilter"
 	"github.com/SigNoz/signoz/pkg/modules/quickfilter/implquickfilter"
 	"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
@@ -34,6 +36,7 @@ type Handlers struct {
 	SpanPercentile  spanpercentile.Handler
 	Services        services.Handler
 	MetricsExplorer metricsexplorer.Handler
+	Promote         promote.Handler
 }
 
 func NewHandlers(modules Modules, providerSettings factory.ProviderSettings, querier querier.Querier, licensing licensing.Licensing) Handlers {
@@ -47,5 +50,6 @@ func NewHandlers(modules Modules, providerSettings factory.ProviderSettings, que
 		Services:        implservices.NewHandler(modules.Services),
 		MetricsExplorer: implmetricsexplorer.NewHandler(modules.MetricsExplorer),
 		SpanPercentile:  implspanpercentile.NewHandler(modules.SpanPercentile),
+		Promote:         implpromote.NewHandler(modules.Promote),
 	}
 }

pkg/signoz/module.go

@@ -20,6 +20,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/organization/implorganization"
 	"github.com/SigNoz/signoz/pkg/modules/preference"
 	"github.com/SigNoz/signoz/pkg/modules/preference/implpreference"
+	"github.com/SigNoz/signoz/pkg/modules/promote"
+	"github.com/SigNoz/signoz/pkg/modules/promote/implpromote"
 	"github.com/SigNoz/signoz/pkg/modules/quickfilter"
 	"github.com/SigNoz/signoz/pkg/modules/quickfilter/implquickfilter"
 	"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
@@ -63,6 +65,7 @@ type Modules struct {
 	Services        services.Module
 	SpanPercentile  spanpercentile.Module
 	MetricsExplorer metricsexplorer.Module
+	Promote         promote.Module
 }
 
 func NewModules(
@@ -103,5 +106,6 @@ func NewModules(
 		SpanPercentile:  implspanpercentile.NewModule(querier, providerSettings),
 		Services:        implservices.NewModule(querier, telemetryStore),
 		MetricsExplorer: implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, providerSettings, config.MetricsExplorer),
-	}
+		Promote:         implpromote.NewModule(telemetryMetadataStore, telemetryStore),
+  }
 }

pkg/telemetrymetadata/body_json_metadata.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
 	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz-otel-collector/constants"
@@ -15,7 +14,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
-	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/huandu/go-sqlbuilder"
 )
@@ -34,6 +32,10 @@ var (
 	CodeFailScanVariant         = errors.MustNewCode("fail_scan_variant")
 	CodeFailBuildJSONPathsQuery = errors.MustNewCode("fail_build_json_paths_query")
 	CodeNoPathsToQueryIndexes   = errors.MustNewCode("no_paths_to_query_indexes_provided")
+
+	CodeFailedToPrepareBatch = errors.MustNewCode("failed_to_prepare_batch_promoted_paths")
+	CodeFailedToSendBatch    = errors.MustNewCode("failed_to_send_batch_promoted_paths")
+	CodeFailedToAppendPath   = errors.MustNewCode("failed_to_append_path_promoted_paths")
 )
 
 // GetBodyJSONPaths extracts body JSON paths from the path_types table
@@ -48,7 +50,7 @@ var (
 // TODO(Piyush): Remove this lint skip
 //
 // nolint:unused
-func getBodyJSONPaths(ctx context.Context, telemetryStore telemetrystore.TelemetryStore,
+func (t *telemetryMetaStore) getBodyJSONPaths(ctx context.Context,
 	fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
 
 	query, args, limit, err := buildGetBodyJSONPathsQuery(fieldKeySelectors)
@@ -56,7 +58,7 @@ func getBodyJSONPaths(ctx context.Context, telemetryStore telemetrystore.Telemet
 		return nil, false, err
 	}
 
-	rows, err := telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
 		return nil, false, errors.WrapInternalf(err, CodeFailExtractBodyJSONKeys, "failed to extract body JSON keys")
 	}
@@ -96,12 +98,12 @@ func getBodyJSONPaths(ctx context.Context, telemetryStore telemetrystore.Telemet
 		return nil, false, errors.WrapInternalf(rows.Err(), CodeFailIterateBodyJSONKeys, "error iterating body JSON keys")
 	}
 
-	promoted, err := GetPromotedPaths(ctx, telemetryStore.ClickhouseDB(), paths...)
+	promoted, err := t.GetPromotedPaths(ctx, paths...)
 	if err != nil {
 		return nil, false, err
 	}
 
-	indexes, err := getJSONPathIndexes(ctx, telemetryStore, paths...)
+	indexes, err := t.getJSONPathIndexes(ctx, paths...)
 	if err != nil {
 		return nil, false, err
 	}
@@ -163,7 +165,7 @@ func buildGetBodyJSONPathsQuery(fieldKeySelectors []*telemetrytypes.FieldKeySele
 // TODO(Piyush): Remove this lint skip
 //
 // nolint:unused
-func getJSONPathIndexes(ctx context.Context, telemetryStore telemetrystore.TelemetryStore, paths ...string) (map[string][]telemetrytypes.JSONDataTypeIndex, error) {
+func (t *telemetryMetaStore) getJSONPathIndexes(ctx context.Context, paths ...string) (map[string][]telemetrytypes.JSONDataTypeIndex, error) {
 	filteredPaths := []string{}
 	for _, path := range paths {
 		if strings.Contains(path, telemetrylogs.ArraySep) || strings.Contains(path, telemetrylogs.ArrayAnyIndex) {
@@ -176,7 +178,7 @@ func getJSONPathIndexes(ctx context.Context, telemetryStore telemetrystore.Telem
 	}
 
 	// list indexes for the paths
-	indexesMap, err := ListLogsJSONIndexes(ctx, telemetryStore, filteredPaths...)
+	indexesMap, err := t.ListLogsJSONIndexes(ctx, filteredPaths...)
 	if err != nil {
 		return nil, errors.WrapInternalf(err, CodeFailLoadLogsJSONIndexes, "failed to list JSON path indexes")
 	}
@@ -215,7 +217,6 @@ func getJSONPathIndexes(ctx context.Context, telemetryStore telemetrystore.Telem
 }
 
 func buildListLogsJSONIndexesQuery(cluster string, filters ...string) (string, []any) {
-	// This aggregates all types per path and gets the max last_seen, then applies LIMIT
 	sb := sqlbuilder.Select(
 		"name", "type_full", "expr", "granularity",
 	).From(fmt.Sprintf("clusterAllReplicas('%s', %s)", cluster, SkipIndexTableName))
@@ -236,9 +237,9 @@ func buildListLogsJSONIndexesQuery(cluster string, filters ...string) (string, [
 	return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
 }
 
-func ListLogsJSONIndexes(ctx context.Context, telemetryStore telemetrystore.TelemetryStore, filters ...string) (map[string][]schemamigrator.Index, error) {
-	query, args := buildListLogsJSONIndexesQuery(telemetryStore.Cluster(), filters...)
-	rows, err := telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+func (t *telemetryMetaStore) ListLogsJSONIndexes(ctx context.Context, filters ...string) (map[string][]schemamigrator.Index, error) {
+	query, args := buildListLogsJSONIndexesQuery(t.telemetrystore.Cluster(), filters...)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
 		return nil, errors.WrapInternalf(err, CodeFailLoadLogsJSONIndexes, "failed to load string indexed columns")
 	}
@@ -264,9 +265,16 @@ func ListLogsJSONIndexes(ctx context.Context, telemetryStore telemetrystore.Tele
 	return indexesMap, nil
 }
 
-func ListPromotedPaths(ctx context.Context, conn clickhouse.Conn) (map[string]struct{}, error) {
-	query := fmt.Sprintf("SELECT path FROM %s.%s", DBName, PromotedPathsTableName)
-	rows, err := conn.Query(ctx, query)
+func (t *telemetryMetaStore) ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error) {
+	sb := sqlbuilder.Select("path").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
+	pathConditions := []string{}
+	for _, path := range paths {
+		pathConditions = append(pathConditions, sb.Equal("path", path))
+	}
+	sb.Where(sb.Or(pathConditions...))
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
 		return nil, errors.WrapInternalf(err, CodeFailLoadPromotedPaths, "failed to load promoted paths")
 	}
@@ -285,14 +293,14 @@ func ListPromotedPaths(ctx context.Context, conn clickhouse.Conn) (map[string]st
 }
 
 // TODO(Piyush): Remove this if not used in future
-func ListJSONValues(ctx context.Context, conn clickhouse.Conn, path string, limit int) (*telemetrytypes.TelemetryFieldValues, bool, error) {
+func (t *telemetryMetaStore) ListJSONValues(ctx context.Context, path string, limit int) (*telemetrytypes.TelemetryFieldValues, bool, error) {
 	path = CleanPathPrefixes(path)
 
 	if strings.Contains(path, telemetrylogs.ArraySep) || strings.Contains(path, telemetrylogs.ArrayAnyIndex) {
 		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "array paths are not supported")
 	}
 
-	promoted, err := IsPathPromoted(ctx, conn, path)
+	promoted, err := t.IsPathPromoted(ctx, path)
 	if err != nil {
 		return nil, false, err
 	}
@@ -325,7 +333,7 @@ func ListJSONValues(ctx context.Context, conn clickhouse.Conn, path string, limi
 	contextWithTimeout, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-	rows, err := conn.Query(contextWithTimeout, query, args...)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(contextWithTimeout, query, args...)
 	if err != nil {
 		if errors.Is(err, context.DeadlineExceeded) {
 			return nil, false, errors.WrapTimeoutf(err, errors.CodeTimeout, "query timed out").WithAdditional("failed to list JSON values")
@@ -447,10 +455,10 @@ func derefValue(v any) any {
 }
 
 // IsPathPromoted checks if a specific path is promoted
-func IsPathPromoted(ctx context.Context, conn clickhouse.Conn, path string) (bool, error) {
+func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (bool, error) {
 	split := strings.Split(path, telemetrylogs.ArraySep)
 	query := fmt.Sprintf("SELECT 1 FROM %s.%s WHERE path = ? LIMIT 1", DBName, PromotedPathsTableName)
-	rows, err := conn.Query(ctx, query, split[0])
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, split[0])
 	if err != nil {
 		return false, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to check if path %s is promoted", path)
 	}
@@ -460,7 +468,7 @@ func IsPathPromoted(ctx context.Context, conn clickhouse.Conn, path string) (boo
 }
 
 // GetPromotedPaths checks if a specific path is promoted
-func GetPromotedPaths(ctx context.Context, conn clickhouse.Conn, paths ...string) (*utils.ConcurrentSet[string], error) {
+func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...string) (*utils.ConcurrentSet[string], error) {
 	sb := sqlbuilder.Select("path").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
 	pathConditions := []string{}
 	for _, path := range paths {
@@ -469,7 +477,7 @@ func GetPromotedPaths(ctx context.Context, conn clickhouse.Conn, paths ...string
 	sb.Where(sb.Or(pathConditions...))
 
 	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-	rows, err := conn.Query(ctx, query, args...)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
 		return nil, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to get promoted paths")
 	}
@@ -494,3 +502,29 @@ func CleanPathPrefixes(path string) string {
 	path = strings.TrimPrefix(path, telemetrylogs.BodyPromotedColumnPrefix)
 	return path
 }
+
+func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string) error {
+	batch, err := t.telemetrystore.ClickhouseDB().PrepareBatch(ctx,
+		fmt.Sprintf("INSERT INTO %s.%s (path, created_at) VALUES", DBName,
+			PromotedPathsTableName))
+	if err != nil {
+		return errors.WrapInternalf(err, CodeFailedToPrepareBatch, "failed to prepare batch")
+	}
+
+	nowMs := uint64(time.Now().UnixMilli())
+	for _, p := range paths {
+		trimmed := strings.TrimSpace(p)
+		if trimmed == "" {
+			continue
+		}
+		if err := batch.Append(trimmed, nowMs); err != nil {
+			_ = batch.Abort()
+			return errors.WrapInternalf(err, CodeFailedToAppendPath, "failed to append path")
+		}
+	}
+
+	if err := batch.Send(); err != nil {
+		return errors.WrapInternalf(err, CodeFailedToSendBatch, "failed to send batch")
+	}
+	return nil
+}

pkg/types/promotetypes/types.go

@@ -0,0 +1,76 @@
+package promotetypes
+
+import (
+	"strings"
+
+	"github.com/SigNoz/signoz-otel-collector/constants"
+	"github.com/SigNoz/signoz-otel-collector/pkg/keycheck"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/telemetrylogs"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+type WrappedIndex struct {
+	JSONDataType telemetrytypes.JSONDataType `json:"-"`
+	ColumnType   string                      `json:"column_type"`
+	Type         string                      `json:"type"`
+	Granularity  int                         `json:"granularity"`
+}
+
+type PromotePath struct {
+	Path    string `json:"path"`
+	Promote bool   `json:"promote,omitempty"`
+
+	Indexes []WrappedIndex `json:"indexes,omitempty"`
+}
+
+func (i *PromotePath) Validate() error {
+	if i.Path == "" {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path is required")
+	}
+
+	if strings.Contains(i.Path, " ") {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path cannot contain spaces")
+	}
+
+	if strings.Contains(i.Path, telemetrylogs.ArraySep) || strings.Contains(i.Path, telemetrylogs.ArrayAnyIndex) {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "array paths can not be promoted or indexed")
+	}
+
+	if strings.HasPrefix(i.Path, constants.BodyJSONColumnPrefix) || strings.HasPrefix(i.Path, constants.BodyPromotedColumnPrefix) {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "`%s`, `%s` don't add these prefixes to the path", constants.BodyJSONColumnPrefix, constants.BodyPromotedColumnPrefix)
+	}
+
+	if !strings.HasPrefix(i.Path, telemetrylogs.BodyJSONStringSearchPrefix) {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path must start with `body.`")
+	}
+
+	// remove the "body." prefix from the path
+	i.Path = strings.TrimPrefix(i.Path, telemetrylogs.BodyJSONStringSearchPrefix)
+
+	isCardinal := keycheck.IsCardinal(i.Path)
+	if isCardinal {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "cardinal paths can not be promoted or indexed")
+	}
+
+	for idx, index := range i.Indexes {
+		if index.Type == "" {
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "index type is required")
+		}
+		if index.Granularity <= 0 {
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "index granularity must be greater than 0")
+		}
+
+		jsonDataType, ok := telemetrytypes.MappingStringToJSONDataType[index.ColumnType]
+		if !ok {
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "invalid column type: %s", index.ColumnType)
+		}
+		if !jsonDataType.IndexSupported {
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "index is not supported for column type: %s", index.ColumnType)
+		}
+
+		i.Indexes[idx].JSONDataType = jsonDataType
+	}
+
+	return nil
+}

pkg/types/telemetrytypes/store.go

@@ -3,6 +3,7 @@ package telemetrytypes
 import (
 	"context"
 
+	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 )
 
@@ -30,4 +31,13 @@ type MetadataStore interface {
 
 	// FetchTemporalityMulti fetches the temporality for multiple metrics
 	FetchTemporalityMulti(ctx context.Context, metricNames ...string) (map[string]metrictypes.Temporality, error)
+
+	// ListLogsJSONIndexes lists the JSON indexes for the logs table.
+	ListLogsJSONIndexes(ctx context.Context, filters ...string) (map[string][]schemamigrator.Index, error)
+
+	// ListPromotedPaths lists the promoted paths.
+	ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error)
+
+	// PromotePaths promotes the paths.
+	PromotePaths(ctx context.Context, paths ...string) error
 }

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"strings"
 
+	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/types/metrictypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
@@ -11,19 +12,23 @@ import (
 // MockMetadataStore implements the MetadataStore interface for testing purposes
 type MockMetadataStore struct {
 	// Maps to store test data
-	KeysMap          map[string][]*telemetrytypes.TelemetryFieldKey
-	RelatedValuesMap map[string][]string
-	AllValuesMap     map[string]*telemetrytypes.TelemetryFieldValues
-	TemporalityMap   map[string]metrictypes.Temporality
+	KeysMap            map[string][]*telemetrytypes.TelemetryFieldKey
+	RelatedValuesMap   map[string][]string
+	AllValuesMap       map[string]*telemetrytypes.TelemetryFieldValues
+	TemporalityMap     map[string]metrictypes.Temporality
+	PromotedPathsMap   map[string]struct{}
+	LogsJSONIndexesMap map[string][]schemamigrator.Index
 }
 
 // NewMockMetadataStore creates a new instance of MockMetadataStore with initialized maps
 func NewMockMetadataStore() *MockMetadataStore {
 	return &MockMetadataStore{
-		KeysMap:          make(map[string][]*telemetrytypes.TelemetryFieldKey),
-		RelatedValuesMap: make(map[string][]string),
-		AllValuesMap:     make(map[string]*telemetrytypes.TelemetryFieldValues),
-		TemporalityMap:   make(map[string]metrictypes.Temporality),
+		KeysMap:            make(map[string][]*telemetrytypes.TelemetryFieldKey),
+		RelatedValuesMap:   make(map[string][]string),
+		AllValuesMap:       make(map[string]*telemetrytypes.TelemetryFieldValues),
+		TemporalityMap:     make(map[string]metrictypes.Temporality),
+		PromotedPathsMap:   make(map[string]struct{}),
+		LogsJSONIndexesMap: make(map[string][]schemamigrator.Index),
 	}
 }
 
@@ -284,3 +289,21 @@ func (m *MockMetadataStore) FetchTemporalityMulti(ctx context.Context, metricNam
 func (m *MockMetadataStore) SetTemporality(metricName string, temporality metrictypes.Temporality) {
 	m.TemporalityMap[metricName] = temporality
 }
+
+// PromotePaths promotes the paths.
+func (m *MockMetadataStore) PromotePaths(ctx context.Context, paths ...string) error {
+	for _, path := range paths {
+		m.PromotedPathsMap[path] = struct{}{}
+	}
+	return nil
+}
+
+// ListPromotedPaths lists the promoted paths.
+func (m *MockMetadataStore) ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error) {
+	return m.PromotedPathsMap, nil
+}
+
+// ListLogsJSONIndexes lists the JSON indexes for the logs table.
+func (m *MockMetadataStore) ListLogsJSONIndexes(ctx context.Context, filters ...string) (map[string][]schemamigrator.Index, error) {
+	return m.LogsJSONIndexesMap, nil
+}