fix: add backticks everywhere

.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
 ##############################################################

ee/query-service/constants/constants.go

@@ -9,6 +9,7 @@ var LicenseAPIKey = GetOrDefaultEnv("SIGNOZ_LICENSE_API_KEY", "")
 var SaasSegmentKey = GetOrDefaultEnv("SIGNOZ_SAAS_SEGMENT_KEY", "")
 var FetchFeatures = GetOrDefaultEnv("FETCH_FEATURES", "false")
 var ZeusFeaturesURL = GetOrDefaultEnv("ZEUS_FEATURES_URL", "ZeusFeaturesURL")
+var BodyJSONQueryEnabled = GetOrDefaultEnv("BODY_JSON_QUERY_ENABLED", "false") == "true"
 
 func GetOrDefaultEnv(key string, fallback string) string {
 	v := os.Getenv(key)

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.40.1
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/SigNoz/govaluate v0.0.0-20240203125216-988004ccc7fd
-	github.com/SigNoz/signoz-otel-collector v0.129.4
+	github.com/SigNoz/signoz-otel-collector v0.129.10-rc.7
 	github.com/antlr4-go/antlr/v4 v4.13.1
 	github.com/antonmedv/expr v1.15.3
 	github.com/cespare/xxhash/v2 v2.3.0
@@ -86,12 +86,19 @@ require (
 )
 
 require (
+	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic v1.14.1 // indirect
+	github.com/bytedance/sonic/loader v0.3.0 // indirect
+	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
 	github.com/redis/go-redis/extra/rediscmd/v9 v9.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/uptrace/opentelemetry-go-extra/otelsql v0.3.2 // indirect
+	go.opentelemetry.io/collector/config/configretry v1.34.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect
 	modernc.org/libc v1.66.10 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect

go.sum

@@ -106,8 +106,8 @@ github.com/SigNoz/expr v1.17.7-beta h1:FyZkleM5dTQ0O6muQfwGpoH5A2ohmN/XTasRCO72g
 github.com/SigNoz/expr v1.17.7-beta/go.mod h1:8/vRC7+7HBzESEqt5kKpYXxrxkr31SaO8r40VO/1IT4=
 github.com/SigNoz/govaluate v0.0.0-20240203125216-988004ccc7fd h1:Bk43AsDYe0fhkbj57eGXx8H3ZJ4zhmQXBnrW523ktj8=
 github.com/SigNoz/govaluate v0.0.0-20240203125216-988004ccc7fd/go.mod h1:nxRcH/OEdM8QxzH37xkGzomr1O0JpYBRS6pwjsWW6Pc=
-github.com/SigNoz/signoz-otel-collector v0.129.4 h1:DGDu9y1I1FU+HX4eECPGmfhnXE4ys4yr7LL6znbf6to=
-github.com/SigNoz/signoz-otel-collector v0.129.4/go.mod h1:xyR+coBzzO04p6Eu+ql2RVYUl/jFD+8hD9lArcc9U7g=
+github.com/SigNoz/signoz-otel-collector v0.129.10-rc.7 h1:r8/+t3ARWek9+X5aH05qavdA9ATbkssfssHh/zjzsEM=
+github.com/SigNoz/signoz-otel-collector v0.129.10-rc.7/go.mod h1:4eJCRUd/P4OiCHXvGYZK8q6oyBVGJFVj/G6qKSoN/TQ=
 github.com/Yiling-J/theine-go v0.6.2 h1:1GeoXeQ0O0AUkiwj2S9Jc0Mzx+hpqzmqsJ4kIC4M9AY=
 github.com/Yiling-J/theine-go v0.6.2/go.mod h1:08QpMa5JZ2pKN+UJCRrCasWYO1IKCdl54Xa836rpmDU=
 github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5/go.mod h1:SkGFH1ia65gfNATL8TAiHDNxPzPdmEL5uirI2Uyuz6c=
@@ -162,6 +162,12 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=
+github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
+github.com/bytedance/sonic v1.14.1 h1:FBMC0zVz5XUmE4z9wF4Jey0An5FueFvOsTKKKtwIl7w=
+github.com/bytedance/sonic v1.14.1/go.mod h1:gi6uhQLMbTdeP0muCnrjHLeCUPyb70ujhnNlhOylAFc=
+github.com/bytedance/sonic/loader v0.3.0 h1:dskwH8edlzNMctoruo8FPTJDF3vLtDT0sXZwvZJyqeA=
+github.com/bytedance/sonic/loader v0.3.0/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFosQU6FxH2JmUe6VI=
 github.com/cactus/go-statsd-client/statsd v0.0.0-20200423205355-cb0885a1018c/go.mod h1:l/bIBLeOl9eX+wxJAzxS4TveKRtAqlyDpHjhkfO0MEI=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
@@ -178,6 +184,8 @@ github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMn
 github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag=
 github.com/circonus-labs/circonusllhist v0.1.3/go.mod h1:kMXHVDlOchFAehlya5ePtbp5jckzBHf4XRpQvBOLI+I=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M=
+github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
@@ -991,6 +999,8 @@ github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GH
 github.com/trivago/tgo v1.0.7 h1:uaWH/XIy9aWYWpjm2CU3RpcqZXmX2ysQ9/Go+d9gyrM=
 github.com/trivago/tgo v1.0.7/go.mod h1:w4dpD+3tzNIIiIfkWWa85w5/B77tlvdZckQ+6PkFnhc=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
+github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
+github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/uptrace/bun v1.2.9 h1:OOt2DlIcRUMSZPr6iXDFg/LaQd59kOxbAjpIVHddKRs=
 github.com/uptrace/bun v1.2.9/go.mod h1:r2ZaaGs9Ru5bpGTr8GQfp8jp+TlCav9grYCPOu2CJSg=
 github.com/uptrace/bun/dialect/pgdialect v1.2.9 h1:caf5uFbOGiXvadV6pA5gn87k0awFFxL1kuuY3SpxnWk=
@@ -1235,6 +1245,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
+golang.org/x/arch v0.0.0-20210923205945-b76863e36670 h1:18EFjUmQOcUvxNYSkA6jO9VAiXCnxFY6NyDX0bHDmkU=
+golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=

pkg/errors/errors.go

@@ -208,3 +208,18 @@ func WrapUnexpectedf(cause error, code Code, format string, args ...any) *base {
 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...)
+}
+
+// NewTimeoutf is a wrapper around Newf with TypeTimeout.
+func NewTimeoutf(code Code, format string, args ...any) *base {
+	return Newf(TypeTimeout, code, format, args...)
+}

pkg/modules/promote/implpromote/handler.go

@@ -0,0 +1,141 @@
+package implpromote
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/http/render"
+	"github.com/SigNoz/signoz/pkg/modules/promote"
+	"github.com/SigNoz/signoz/pkg/telemetrylogs"
+	"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) HandlePromote(w http.ResponseWriter, r *http.Request) {
+	_, err := authtypes.ClaimsFromContext(r.Context())
+	if err != nil {
+		render.Error(w, errors.NewInternalf(errors.CodeInternal, "failed to get org id from context"))
+		return
+	}
+
+	switch r.Method {
+	case http.MethodGet:
+		h.GetPromotedAndIndexedPaths(w, r)
+		return
+	case http.MethodPost:
+		h.PromotePaths(w, r)
+		return
+	case http.MethodDelete:
+		h.DropIndex(w, r)
+		return
+	default:
+		render.Error(w, errors.NewMethodNotAllowedf(errors.CodeMethodNotAllowed, "method not allowed"))
+		return
+	}
+}
+
+func (h *handler) DropIndex(w http.ResponseWriter, r *http.Request) {
+	var req promotetypes.PromotePath
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		render.Error(w, errors.NewInvalidInputf(errors.CodeInvalidInput, "Invalid data"))
+		return
+	}
+
+	err := h.module.DropIndex(r.Context(), req)
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	render.Success(w, http.StatusOK, nil)
+}
+
+func (h *handler) PromotePaths(w http.ResponseWriter, r *http.Request) {
+	var req []promotetypes.PromotePath
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		render.Error(w, errors.NewInvalidInputf(errors.CodeInvalidInput, "Invalid data"))
+		return
+	}
+
+	// Delegate all processing to the reader
+	err := h.module.PromoteAndIndexPaths(r.Context(), req...)
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	render.Success(w, http.StatusOK, nil)
+}
+
+func (h *handler) GetPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request) {
+	response, err := func() ([]promotetypes.PromotePath, error) {
+		indexes, err := h.module.ListBodySkipIndexes(r.Context())
+		if err != nil {
+			return nil, err
+		}
+		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 := h.module.ListPromotedPaths(r.Context())
+		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
+	}()
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+
+	render.Success(w, http.StatusOK, response)
+}

pkg/modules/promote/implpromote/module.go

@@ -0,0 +1,238 @@
+package implpromote
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
+	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/telemetrymetadata"
+	"github.com/SigNoz/signoz/pkg/telemetrystore"
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+	"github.com/huandu/go-sqlbuilder"
+)
+
+var (
+	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")
+	CodeFailedToCreateIndex        = errors.MustNewCode("failed_to_create_index_promoted_paths")
+	CodeFailedToDropIndex          = errors.MustNewCode("failed_to_drop_index_promoted_paths")
+	CodeFailedToQueryPromotedPaths = errors.MustNewCode("failed_to_query_promoted_paths")
+)
+
+type module struct {
+	store telemetrystore.TelemetryStore
+}
+
+func NewModule(store telemetrystore.TelemetryStore) promote.Module {
+	return &module{store: store}
+}
+
+func (m *module) ListBodySkipIndexes(ctx context.Context) ([]schemamigrator.Index, error) {
+	return telemetrymetadata.ListLogsJSONIndexes(ctx, m.store)
+}
+
+func (m *module) ListPromotedPaths(ctx context.Context) ([]string, error) {
+	paths, err := telemetrymetadata.ListPromotedPaths(ctx, m.store.ClickhouseDB())
+	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")
+	}
+
+	batch, err := m.store.ClickhouseDB().PrepareBatch(ctx,
+		fmt.Sprintf("INSERT INTO %s.%s (path, created_at) VALUES", telemetrymetadata.DBName,
+			telemetrymetadata.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
+}
+
+// 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.store.Cluster())
+		if err := m.store.ClickhouseDB().Exec(ctx, op.ToSQL()); err != nil {
+			return errors.WrapInternalf(err, CodeFailedToCreateIndex, "failed to create index")
+		}
+	}
+
+	return nil
+}
+
+func (m *module) DropIndex(ctx context.Context, path promotetypes.PromotePath) error {
+	// validate the paths
+	if err := path.Validate(); err != nil {
+		return err
+	}
+
+	promoted, err := telemetrymetadata.IsPathPromoted(ctx, m.store.ClickhouseDB(), path.Path)
+	if err != nil {
+		return err
+	}
+	parentColumn := telemetrylogs.LogsV2BodyJSONColumn
+	if promoted {
+		parentColumn = telemetrylogs.LogsV2BodyPromotedColumn
+	}
+
+	for _, index := range path.Indexes {
+		typeIndex := schemamigrator.IndexTypeTokenBF
+		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)
+		}
+
+		alterStmt := schemamigrator.AlterTableDropIndex{
+			Database: telemetrylogs.DBName,
+			Table:    telemetrylogs.LogsV2LocalTableName,
+			Index: schemamigrator.Index{
+				Name:        schemamigrator.JSONSubColumnIndexName(parentColumn, path.Path, index.JSONDataType.StringValue(), typeIndex),
+				Expression:  schemamigrator.JSONSubColumnIndexExpr(parentColumn, path.Path, index.JSONDataType.StringValue()),
+				Type:        index.Type,
+				Granularity: index.Granularity,
+			},
+		}
+		op := alterStmt.OnCluster(m.store.Cluster())
+		if err := m.store.ClickhouseDB().Exec(ctx, op.ToSQL()); err != nil {
+			return errors.WrapInternalf(err, CodeFailedToDropIndex, "failed to drop 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")
+	}
+
+	// validate the paths
+	for _, path := range paths {
+		if err := path.Validate(); err != nil {
+			return err
+		}
+	}
+
+	sb := sqlbuilder.NewSelectBuilder().From(fmt.Sprintf("%s.%s", telemetrymetadata.DBName, telemetrymetadata.PromotedPathsTableName)).Select("path")
+	cond := []string{}
+	for _, path := range paths {
+		cond = append(cond, sb.Equal("path", path.Path))
+	}
+	sb.Where(sb.Or(cond...))
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	rows, err := m.store.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return errors.WrapInternalf(err, CodeFailedToQueryPromotedPaths, "failed to query promoted paths")
+	}
+	defer rows.Close()
+
+	// Load existing promoted paths once
+	existingPromotedPaths := make(map[string]struct{})
+	for rows.Next() {
+		var p string
+		if err := rows.Scan(&p); err == nil {
+			existingPromotedPaths[p] = struct{}{}
+		}
+	}
+
+	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 {
+				typeIndex := schemamigrator.IndexTypeTokenBF
+				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,20 @@
+package promote
+
+import (
+	"context"
+	"net/http"
+
+	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+)
+
+type Module interface {
+	ListBodySkipIndexes(ctx context.Context) ([]schemamigrator.Index, error)
+	ListPromotedPaths(ctx context.Context) ([]string, error)
+	PromoteAndIndexPaths(ctx context.Context, paths ...promotetypes.PromotePath) error
+	DropIndex(ctx context.Context, path promotetypes.PromotePath) error
+}
+
+type Handler interface {
+	HandlePromote(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

@@ -549,6 +549,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)
 
@@ -4020,6 +4021,9 @@ 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.HandlePromote)).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 }
 
 func (aH *APIHandler) logFields(w http.ResponseWriter, r *http.Request) {

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

@@ -404,7 +404,7 @@ func buildLogsQuery(panelType v3.PanelType, start, end, step int64, mq *v3.Build
 	// if noop create the query and return
 	if mq.AggregateOperator == v3.AggregateOperatorNoOp {
 		// with noop any filter or different order by other than ts will use new table
-		sqlSelect := constants.LogsSQLSelectV2
+		sqlSelect := constants.LogsSQLSelectV2()
 		queryTmpl := sqlSelect + "from signoz_logs.%s where %s%s order by %s"
 		query := fmt.Sprintf(queryTmpl, DISTRIBUTED_LOGS_V2, timeFilter, filterSubQuery, orderBy)
 		return query, nil
@@ -488,7 +488,7 @@ func buildLogsLiveTailQuery(mq *v3.BuilderQuery) (string, error) {
 	// the reader will add the timestamp and id filters
 	switch mq.AggregateOperator {
 	case v3.AggregateOperatorNoOp:
-		query := constants.LogsSQLSelectV2 + "from signoz_logs." + DISTRIBUTED_LOGS_V2 + " where "
+		query := constants.LogsSQLSelectV2() + "from signoz_logs." + DISTRIBUTED_LOGS_V2 + " where "
 		if len(filterSubQuery) > 0 {
 			query = query + filterSubQuery + " AND "
 		}

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/query-service/constants/constants.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/query-service/model"
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
+	"github.com/huandu/go-sqlbuilder"
 )
 
 const (
@@ -216,13 +217,6 @@ const (
 		"CAST((attributes_bool_key, attributes_bool_value), 'Map(String, Bool)') as  attributes_bool," +
 		"CAST((resources_string_key, resources_string_value), 'Map(String, String)') as resources_string," +
 		"CAST((scope_string_key, scope_string_value), 'Map(String, String)') as scope "
-	LogsSQLSelectV2 = "SELECT " +
-		"timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, " +
-		"attributes_string, " +
-		"attributes_number, " +
-		"attributes_bool, " +
-		"resources_string, " +
-		"scope_string "
 	TracesExplorerViewSQLSelectWithSubQuery = "(SELECT traceID, durationNano, " +
 		"serviceName, name FROM %s.%s WHERE parentSpanID = '' AND %s ORDER BY durationNano DESC LIMIT 1 BY traceID"
 	TracesExplorerViewSQLSelectBeforeSubQuery = "SELECT subQuery.serviceName as `subQuery.serviceName`, subQuery.name as `subQuery.name`, count() AS " +
@@ -692,6 +686,7 @@ var StaticFieldsTraces = map[string]v3.AttributeKey{}
 var IsDotMetricsEnabled = false
 var PreferSpanMetrics = false
 var MaxJSONFlatteningDepth = 1
+var BodyJSONQueryEnabled = GetOrDefaultEnv("BODY_JSON_QUERY_ENABLED", "false") == "true"
 
 func init() {
 	StaticFieldsTraces = maps.Clone(NewStaticFieldsTraces)
@@ -732,3 +727,15 @@ const InspectMetricsMaxTimeDiff = 1800000
 
 const DotMetricsEnabled = "DOT_METRICS_ENABLED"
 const maxJSONFlatteningDepth = "MAX_JSON_FLATTENING_DEPTH"
+
+func LogsSQLSelectV2() string {
+	sb := sqlbuilder.NewSelectBuilder()
+	columns := []string{"timestamp", "id", "trace_id", "span_id", "trace_flags", "severity_text", "severity_number", "scope_name", "scope_version", "body"}
+	if BodyJSONQueryEnabled {
+		columns = append(columns, "body_json", "body_json_promoted")
+	}
+	columns = append(columns, "attributes_string", "attributes_number", "attributes_bool", "resources_string", "scope_string")
+	sb.Select(columns...)
+	query, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return query + " " // add space to avoid concatenation issues
+}

pkg/querybuilder/agg_rewrite.go

@@ -198,7 +198,6 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
 				FieldMapper:      v.fieldMapper,
 				ConditionBuilder: v.conditionBuilder,
 				FullTextColumn:   v.fullTextColumn,
-				JsonBodyPrefix:   v.jsonBodyPrefix,
 				JsonKeyToKey:     v.jsonKeyToKey,
             }, 0, 0,
 		)

pkg/querybuilder/query_to_keys_test.go

@@ -7,7 +7,6 @@ import (
 )
 
 func TestQueryToKeys(t *testing.T) {
-
 	testCases := []struct {
 		query        string
 		expectedKeys []telemetrytypes.FieldKeySelector
@@ -66,9 +65,9 @@ func TestQueryToKeys(t *testing.T) {
 			query: `body.user_ids[*] = 123`,
 			expectedKeys: []telemetrytypes.FieldKeySelector{
 				{
-					Name:          "body.user_ids[*]",
+					Name:          "user_ids[*]",
 					Signal:        telemetrytypes.SignalUnspecified,
-					FieldContext:  telemetrytypes.FieldContextUnspecified,
+					FieldContext:  telemetrytypes.FieldContextBody,
 					FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
 				},
 			},

pkg/querybuilder/resourcefilter/statement_builder.go

@@ -162,7 +162,6 @@ func (b *resourceFilterStatementBuilder[T]) addConditions(
 			ConditionBuilder:   b.conditionBuilder,
 			FieldKeys:          keys,
 			FullTextColumn:     b.fullTextColumn,
-			JsonBodyPrefix:     b.jsonBodyPrefix,
 			JsonKeyToKey:       b.jsonKeyToKey,
 			SkipFullTextFilter: true,
 			SkipFunctionCalls:  true,

pkg/querybuilder/where_clause_visitor.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	grammar "github.com/SigNoz/signoz/pkg/parser/grammar"
+	"github.com/SigNoz/signoz/pkg/query-service/constants"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/antlr4-go/antlr/v4"
@@ -33,7 +34,6 @@ type filterExpressionVisitor struct {
 	mainErrorURL       string
 	builder            *sqlbuilder.SelectBuilder
 	fullTextColumn     *telemetrytypes.TelemetryFieldKey
-	jsonBodyPrefix     string
 	jsonKeyToKey       qbtypes.JsonKeyToFieldFunc
 	skipResourceFilter bool
 	skipFullTextFilter bool
@@ -53,7 +53,6 @@ type FilterExprVisitorOpts struct {
 	FieldKeys          map[string][]*telemetrytypes.TelemetryFieldKey
 	Builder            *sqlbuilder.SelectBuilder
 	FullTextColumn     *telemetrytypes.TelemetryFieldKey
-	JsonBodyPrefix     string
 	JsonKeyToKey       qbtypes.JsonKeyToFieldFunc
 	SkipResourceFilter bool
 	SkipFullTextFilter bool
@@ -73,7 +72,6 @@ func newFilterExpressionVisitor(opts FilterExprVisitorOpts) *filterExpressionVis
 		fieldKeys:          opts.FieldKeys,
 		builder:            opts.Builder,
 		fullTextColumn:     opts.FullTextColumn,
-		jsonBodyPrefix:     opts.JsonBodyPrefix,
 		jsonKeyToKey:       opts.JsonKeyToKey,
 		skipResourceFilter: opts.SkipResourceFilter,
 		skipFullTextFilter: opts.SkipFullTextFilter,
@@ -172,7 +170,7 @@ func PrepareWhereClause(query string, opts FilterExprVisitorOpts, startNs uint64
 
 	whereClause := sqlbuilder.NewWhereClause().AddWhereExpr(visitor.builder.Args, cond)
 
-	return &PreparedWhereClause{whereClause, visitor.warnings, visitor.mainWarnURL}, nil
+	return &PreparedWhereClause{WhereClause: whereClause, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL}, nil
 }
 
 // Visit dispatches to the specific visit method based on node type
@@ -717,7 +715,7 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
 			conds = append(conds, fmt.Sprintf("hasToken(LOWER(%s), LOWER(%s))", key.Name, v.builder.Var(value[0])))
 		} else {
 			// this is that all other functions only support array fields
-			if strings.HasPrefix(key.Name, v.jsonBodyPrefix) {
+			if key.FieldContext == telemetrytypes.FieldContextBody {
 				fieldName, _ = v.jsonKeyToKey(context.Background(), key, qbtypes.FilterOperatorUnknown, value)
 			} else {
 				// TODO(add docs for json body search)
@@ -808,10 +806,8 @@ func (v *filterExpressionVisitor) VisitValue(ctx *grammar.ValueContext) any {
 
 // VisitKey handles field/column references
 func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
-
 	fieldKey := telemetrytypes.GetFieldKeyFromKeyText(ctx.GetText())
-
-	keyName := strings.TrimPrefix(fieldKey.Name, v.jsonBodyPrefix)
+	keyName := fieldKey.Name
 
 	fieldKeysForName := v.fieldKeys[keyName]
 
@@ -845,10 +841,11 @@ func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
 	// if there is a field with the same name as attribute/resource attribute
 	// Since it will ORed with the fieldKeysForName, it will not result empty
 	// when either of them have values
-	if strings.HasPrefix(fieldKey.Name, v.jsonBodyPrefix) && v.jsonBodyPrefix != "" {
-		if keyName != "" {
-			fieldKeysForName = append(fieldKeysForName, &fieldKey)
-		}
+	// Note: Skip this logic if body json query is enabled so we can look up the key inside fields
+	//
+	// TODO(Piyush): After entire migration this is supposed to be removed.
+	if !constants.BodyJSONQueryEnabled && fieldKey.FieldContext == telemetrytypes.FieldContextBody {
+		fieldKeysForName = append(fieldKeysForName, &fieldKey)
 	}
 
 	if len(fieldKeysForName) == 0 {
@@ -859,7 +856,7 @@ func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
 			return v.fieldKeys[keyWithContext]
 		}
 
-		if strings.HasPrefix(fieldKey.Name, v.jsonBodyPrefix) && v.jsonBodyPrefix != "" && keyName == "" {
+		if fieldKey.FieldContext == telemetrytypes.FieldContextBody && keyName == "" {
 			v.errors = append(v.errors, "missing key for body json search - expected key of the form `body.key` (ex: `body.status`)")
 		} else if !v.ignoreNotFoundKeys {
 			// TODO(srikanthccv): do we want to return an error here?

pkg/signoz/handler.go

@@ -13,6 +13,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"
@@ -46,6 +48,7 @@ type Handlers struct {
 	Session        session.Handler
 	SpanPercentile spanpercentile.Handler
 	Services       services.Handler
+	Promote        promote.Handler
 }
 
 func NewHandlers(modules Modules, providerSettings factory.ProviderSettings, querier querier.Querier, licensing licensing.Licensing) Handlers {
@@ -63,5 +66,6 @@ func NewHandlers(modules Modules, providerSettings factory.ProviderSettings, que
 		Session:        implsession.NewHandler(modules.Session),
 		Services:       implservices.NewHandler(modules.Services),
 		SpanPercentile: implspanpercentile.NewHandler(modules.SpanPercentile),
+		Promote:        implpromote.NewHandler(modules.Promote),
 	}
 }

pkg/signoz/module.go

@@ -17,6 +17,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"
@@ -58,6 +60,7 @@ type Modules struct {
 	Session        session.Module
 	Services       services.Module
 	SpanPercentile spanpercentile.Module
+	Promote        promote.Module
 }
 
 func NewModules(
@@ -94,5 +97,6 @@ func NewModules(
 		Session:        implsession.NewModule(providerSettings, authNs, user, userGetter, implauthdomain.NewModule(implauthdomain.NewStore(sqlstore)), tokenizer, orgGetter),
 		SpanPercentile: implspanpercentile.NewModule(querier, providerSettings),
 		Services:       implservices.NewModule(querier, telemetryStore),
+		Promote:        implpromote.NewModule(telemetryStore),
 	}
 }

pkg/telemetrylogs/condition_builder.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"slices"
-	"strings"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -52,7 +51,8 @@ func (c *conditionBuilder) conditionFor(
 		return "", err
 	}
 
-	if strings.HasPrefix(key.Name, BodyJSONStringSearchPrefix) {
+	// Check if this is a body JSON search - either by FieldContext
+	if key.FieldContext == telemetrytypes.FieldContextBody {
 		tblFieldName, value = GetBodyJSONKey(ctx, key, operator, value)
 	}
 
@@ -156,7 +156,8 @@ func (c *conditionBuilder) conditionFor(
 	// key membership checks, so depending on the column type, the condition changes
 	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
 
-		if strings.HasPrefix(key.Name, BodyJSONStringSearchPrefix) {
+		// Check if this is a body JSON search - by FieldContext
+		if key.FieldContext == telemetrytypes.FieldContextBody {
 			if operator == qbtypes.FilterOperatorExists {
 				return GetBodyJSONKeyForExists(ctx, key, operator, value), nil
 			} else {
@@ -165,13 +166,15 @@ func (c *conditionBuilder) conditionFor(
 		}
 
 		var value any
-		switch column.Type {
-		case schema.JSONColumnType{}:
+		// schema.JSONColumnType{} now can not be used in switch cases, so we need to check if the column is a JSON column
+		if column.IsJSONColumn() {
 			if operator == qbtypes.FilterOperatorExists {
 				return sb.IsNotNull(tblFieldName), nil
 			} else {
 				return sb.IsNull(tblFieldName), nil
 			}
+		}
+		switch column.Type {
 		case schema.ColumnTypeString, schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}:
 			value = ""
 			if operator == qbtypes.FilterOperatorExists {
@@ -218,8 +221,8 @@ func (c *conditionBuilder) ConditionFor(
 	operator qbtypes.FilterOperator,
 	value any,
 	sb *sqlbuilder.SelectBuilder,
-    _ uint64,
-    _ uint64,
+	_ uint64,
+	_ uint64,
 ) (string, error) {
 	condition, err := c.conditionFor(ctx, key, operator, value, sb)
 	if err != nil {
@@ -230,7 +233,7 @@ func (c *conditionBuilder) ConditionFor(
 		// skip adding exists filter for intrinsic fields
 		// with an exception for body json search
 		field, _ := c.fm.FieldFor(ctx, key)
-		if slices.Contains(maps.Keys(IntrinsicFields), field) && !strings.HasPrefix(key.Name, BodyJSONStringSearchPrefix) {
+		if slices.Contains(maps.Keys(IntrinsicFields), field) && key.FieldContext != telemetrytypes.FieldContextBody {
 			return condition, nil
 		}
 

pkg/telemetrylogs/condition_builder_test.go

@@ -276,7 +276,7 @@ func TestConditionFor(t *testing.T) {
 	for _, tc := range testCases {
 		sb := sqlbuilder.NewSelectBuilder()
 		t.Run(tc.name, func(t *testing.T) {
-            cond, err := conditionBuilder.ConditionFor(ctx, &tc.key, tc.operator, tc.value, sb, 0, 0)
+			cond, err := conditionBuilder.ConditionFor(ctx, &tc.key, tc.operator, tc.value, sb, 0, 0)
 			sb.Where(cond)
 
 			if tc.expectedError != nil {
@@ -331,7 +331,7 @@ func TestConditionForMultipleKeys(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			var err error
 			for _, key := range tc.keys {
-                cond, err := conditionBuilder.ConditionFor(ctx, &key, tc.operator, tc.value, sb, 0, 0)
+				cond, err := conditionBuilder.ConditionFor(ctx, &key, tc.operator, tc.value, sb, 0, 0)
 				sb.Where(cond)
 				if err != nil {
 					t.Fatalf("Error getting condition for key %s: %v", key.Name, err)
@@ -363,7 +363,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Equal operator - int64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorEqual,
 			value:         200,
@@ -373,7 +374,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Equal operator - float64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.duration_ms",
+				Name:         "duration_ms",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorEqual,
 			value:         405.5,
@@ -383,7 +385,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Equal operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.method",
+				Name:         "http.method",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorEqual,
 			value:         "GET",
@@ -393,7 +396,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Equal operator - bool",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.success",
+				Name:         "http.success",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorEqual,
 			value:         true,
@@ -403,7 +407,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Exists operator",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorExists,
 			value:         nil,
@@ -413,7 +418,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Not Exists operator",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorNotExists,
 			value:         nil,
@@ -423,7 +429,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Greater than operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorGreaterThan,
 			value:         "200",
@@ -433,7 +440,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Greater than operator - int64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorGreaterThan,
 			value:         200,
@@ -443,7 +451,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Less than operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorLessThan,
 			value:         "300",
@@ -453,7 +462,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Less than operator - int64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorLessThan,
 			value:         300,
@@ -463,7 +473,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Contains operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorContains,
 			value:         "200",
@@ -473,7 +484,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Not Contains operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorNotContains,
 			value:         "200",
@@ -483,7 +495,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Between operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorBetween,
 			value:         []any{"200", "300"},
@@ -493,7 +506,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "Between operator - int64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorBetween,
 			value:         []any{400, 500},
@@ -503,7 +517,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "In operator - string",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorIn,
 			value:         []any{"200", "300"},
@@ -513,7 +528,8 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 		{
 			name: "In operator - int64",
 			key: telemetrytypes.TelemetryFieldKey{
-				Name: "body.http.status_code",
+				Name:         "http.status_code",
+				FieldContext: telemetrytypes.FieldContextBody,
 			},
 			operator:      qbtypes.FilterOperatorIn,
 			value:         []any{401, 404, 500},
@@ -528,7 +544,7 @@ func TestConditionForJSONBodySearch(t *testing.T) {
 	for _, tc := range testCases {
 		sb := sqlbuilder.NewSelectBuilder()
 		t.Run(tc.name, func(t *testing.T) {
-            cond, err := conditionBuilder.ConditionFor(ctx, &tc.key, tc.operator, tc.value, sb, 0, 0)
+			cond, err := conditionBuilder.ConditionFor(ctx, &tc.key, tc.operator, tc.value, sb, 0, 0)
 			sb.Where(cond)
 
 			if tc.expectedError != nil {

pkg/telemetrylogs/const.go

@@ -1,6 +1,8 @@
 package telemetrylogs
 
 import (
+	"github.com/SigNoz/signoz-otel-collector/constants"
+	"github.com/SigNoz/signoz-otel-collector/exporter/jsontypeexporter"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 )
@@ -16,6 +18,8 @@ const (
 	LogsV2TimestampColumn         = "timestamp"
 	LogsV2ObservedTimestampColumn = "observed_timestamp"
 	LogsV2BodyColumn              = "body"
+	LogsV2BodyJSONColumn          = constants.BodyJSONColumn
+	LogsV2BodyPromotedColumn      = constants.BodyPromotedColumn
 	LogsV2TraceIDColumn           = "trace_id"
 	LogsV2SpanIDColumn            = "span_id"
 	LogsV2TraceFlagsColumn        = "trace_flags"
@@ -30,6 +34,11 @@ const (
 	LogsV2AttributesBoolColumn   = "attributes_bool"
 	LogsV2ResourcesStringColumn  = "resources_string"
 	LogsV2ScopeStringColumn      = "scope_string"
+
+	BodyJSONColumnPrefix     = constants.BodyJSONColumnPrefix
+	BodyPromotedColumnPrefix = constants.BodyPromotedColumnPrefix
+	ArraySep                 = jsontypeexporter.ArraySeparator
+	ArrayAnyIndex            = "[*]."
 )
 
 var (

pkg/telemetrylogs/field_mapper.go

@@ -82,10 +82,13 @@ func (m *fieldMapper) getColumn(_ context.Context, key *telemetrytypes.Telemetry
 		case telemetrytypes.FieldDataTypeBool:
 			return logsV2Columns["attributes_bool"], nil
 		}
+	case telemetrytypes.FieldContextBody:
+		// body context fields are stored in the body column
+		return logsV2Columns["body"], nil
 	case telemetrytypes.FieldContextLog, telemetrytypes.FieldContextUnspecified:
 		col, ok := logsV2Columns[key.Name]
 		if !ok {
-			// check if the key has body JSON search
+			// check if the key has body JSON search (backward compatibility)
 			if strings.HasPrefix(key.Name, BodyJSONStringSearchPrefix) {
 				return logsV2Columns["body"], nil
 			}
@@ -103,8 +106,8 @@ func (m *fieldMapper) FieldFor(ctx context.Context, key *telemetrytypes.Telemetr
 		return "", err
 	}
 
-	switch column.Type {
-	case schema.JSONColumnType{}:
+	// schema.JSONColumnType{} now can not be used in switch cases, so we need to check if the column is a JSON column
+	if column.IsJSONColumn() {
 		// json is only supported for resource context as of now
 		if key.FieldContext != telemetrytypes.FieldContextResource {
 			return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns, got %s", key.FieldContext.String)
@@ -121,7 +124,8 @@ func (m *fieldMapper) FieldFor(ctx context.Context, key *telemetrytypes.Telemetr
 		} else {
 			return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, mapContains(%s, '%s'), %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldColumn.Name, key.Name, oldKeyName), nil
 		}
-
+	}
+	switch column.Type {
 	case schema.ColumnTypeString,
 		schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 		schema.ColumnTypeUInt64,

pkg/telemetrylogs/filter_expr_like_warning_test.go

@@ -21,7 +21,6 @@ func TestLikeAndILikeWithoutWildcards_Warns(t *testing.T) {
 		ConditionBuilder: cb,
 		FieldKeys:        keys,
 		FullTextColumn:   DefaultFullTextColumn,
-		JsonBodyPrefix:   BodyJSONStringSearchPrefix,
 		JsonKeyToKey:     GetBodyJSONKey,
 	}
 
@@ -58,7 +57,6 @@ func TestLikeAndILikeWithWildcards_NoWarn(t *testing.T) {
 		ConditionBuilder: cb,
 		FieldKeys:        keys,
 		FullTextColumn:   DefaultFullTextColumn,
-		JsonBodyPrefix:   BodyJSONStringSearchPrefix,
 		JsonKeyToKey:     GetBodyJSONKey,
 	}
 

pkg/telemetrylogs/filter_expr_logs_body_json_test.go

@@ -27,8 +27,7 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
 		FullTextColumn: &telemetrytypes.TelemetryFieldKey{
 			Name: "body",
 		},
-		JsonBodyPrefix: "body",
-		JsonKeyToKey:   GetBodyJSONKey,
+		JsonKeyToKey: GetBodyJSONKey,
 	}
 
 	testCases := []struct {
@@ -163,7 +162,7 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(fmt.Sprintf("%s: %s", tc.category, limitString(tc.query, 50)), func(t *testing.T) {
 
-            clause, err := querybuilder.PrepareWhereClause(tc.query, opts, 0, 0)
+			clause, err := querybuilder.PrepareWhereClause(tc.query, opts, 0, 0)
 
 			if tc.shouldPass {
 				if err != nil {

pkg/telemetrylogs/filter_expr_logs_test.go

@@ -27,7 +27,6 @@ func TestFilterExprLogs(t *testing.T) {
 		ConditionBuilder: cb,
 		FieldKeys:        keys,
 		FullTextColumn:   DefaultFullTextColumn,
-		JsonBodyPrefix:   BodyJSONStringSearchPrefix,
 		JsonKeyToKey:     GetBodyJSONKey,
 	}
 
@@ -2448,7 +2447,6 @@ func TestFilterExprLogsConflictNegation(t *testing.T) {
 		ConditionBuilder: cb,
 		FieldKeys:        keys,
 		FullTextColumn:   DefaultFullTextColumn,
-		JsonBodyPrefix:   BodyJSONStringSearchPrefix,
 		JsonKeyToKey:     GetBodyJSONKey,
 	}
 

pkg/telemetrylogs/json.go

@@ -69,7 +69,7 @@ func inferDataType(value any, operator qbtypes.FilterOperator, key *telemetrytyp
 }
 
 func getBodyJSONPath(key *telemetrytypes.TelemetryFieldKey) string {
-	parts := strings.Split(key.Name, ".")[1:]
+	parts := strings.Split(key.Name, ".")
 	newParts := []string{}
 	for _, part := range parts {
 		if strings.HasSuffix(part, "[*]") {

pkg/telemetrylogs/json_access_pb.go

@@ -0,0 +1,149 @@
+package telemetrylogs
+
+import (
+	"context"
+	"slices"
+	"strings"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+var (
+	CodePlanIndexOutOfBounds = errors.MustNewCode("plan_index_out_of_bounds")
+)
+
+type JSONAccessPlanBuilder struct {
+	key        *telemetrytypes.TelemetryFieldKey
+	value      any
+	op         qbtypes.FilterOperator
+	parts      []string
+	getTypes   func(ctx context.Context, path string) ([]telemetrytypes.JSONDataType, error)
+	isPromoted bool
+}
+
+// buildPlan recursively builds the path plan tree
+func (pb *JSONAccessPlanBuilder) buildPlan(ctx context.Context, index int, parent *telemetrytypes.JSONAccessNode, isDynArrChild bool) (*telemetrytypes.JSONAccessNode, error) {
+	if index >= len(pb.parts) {
+		return nil, errors.NewInvalidInputf(CodePlanIndexOutOfBounds, "index is out of bounds")
+	}
+
+	part := pb.parts[index]
+	pathSoFar := strings.Join(pb.parts[:index+1], ArraySep)
+	isTerminal := index == len(pb.parts)-1
+
+	// Calculate progression parameters based on parent's values
+	var maxTypes, maxPaths int
+	if isDynArrChild {
+		// Child of Dynamic array - reset progression to base values (16, 256)
+		// This happens when we switch from Array(Dynamic) to Array(JSON)
+		maxTypes = 16
+		maxPaths = 256
+	} else if parent != nil {
+		// Child of JSON array - use parent's progression divided by 2 and 4
+		maxTypes = parent.MaxDynamicTypes / 2
+		maxPaths = parent.MaxDynamicPaths / 4
+		if maxTypes < 0 {
+			maxTypes = 0
+		}
+		if maxPaths < 0 {
+			maxPaths = 0
+		}
+	}
+
+	types, err := pb.getTypes(ctx, pathSoFar)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create node for this path segment
+	node := &telemetrytypes.JSONAccessNode{
+		Name:            part,
+		IsTerminal:      isTerminal,
+		AvailableTypes:  types,
+		Branches:        make(map[telemetrytypes.JSONAccessBranchType]*telemetrytypes.JSONAccessNode),
+		Parent:          parent,
+		MaxDynamicTypes: maxTypes,
+		MaxDynamicPaths: maxPaths,
+	}
+
+	hasJSON := slices.Contains(node.AvailableTypes, telemetrytypes.ArrayJSON)
+	hasDynamic := slices.Contains(node.AvailableTypes, telemetrytypes.ArrayDynamic)
+
+	// Configure terminal if this is the last part
+	if isTerminal {
+		valueType, _ := inferDataType(pb.value, pb.op, pb.key)
+		node.TerminalConfig = &telemetrytypes.TerminalConfig{
+			Key:       pb.key,
+			ElemType:  *pb.key.JSONDataType,
+			ValueType: telemetrytypes.MappingFieldDataTypeToJSONDataType[valueType],
+		}
+	} else {
+		if hasJSON {
+			node.Branches[telemetrytypes.BranchJSON], err = pb.buildPlan(ctx, index+1, node, false)
+			if err != nil {
+				return nil, err
+			}
+		}
+		if hasDynamic {
+			node.Branches[telemetrytypes.BranchDynamic], err = pb.buildPlan(ctx, index+1, node, true)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	return node, nil
+}
+
+// PlanJSON builds a tree structure representing the complete JSON path traversal
+// that precomputes all possible branches and their types
+func PlanJSON(ctx context.Context, key *telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator,
+	value any,
+	getTypes func(ctx context.Context, path string) ([]telemetrytypes.JSONDataType, error),
+) (telemetrytypes.JSONAccessPlan, error) {
+	// if path is empty, return nil
+	if key.Name == "" {
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "path is empty")
+	}
+
+	// TODO: PlanJSON requires the Start and End of the Query to select correct column between promoted and body_json using
+	// creation time in distributed_promoted_paths
+	path := strings.ReplaceAll(key.Name, ArrayAnyIndex, ArraySep)
+	parts := strings.Split(path, ArraySep)
+
+	pb := &JSONAccessPlanBuilder{
+		key:        key,
+		op:         op,
+		value:      value,
+		parts:      parts,
+		getTypes:   getTypes,
+		isPromoted: key.Materialized,
+	}
+	plans := telemetrytypes.JSONAccessPlan{}
+
+	node, err := pb.buildPlan(ctx, 0,
+		telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn,
+			32, 0),
+		false,
+	)
+	if err != nil {
+		return nil, err
+	}
+	plans = append(plans, node)
+
+	if pb.isPromoted {
+		node, err := pb.buildPlan(ctx, 0,
+			telemetrytypes.NewRootJSONAccessNode(LogsV2BodyPromotedColumn,
+				32, 1024),
+			true,
+		)
+		if err != nil {
+			return nil, err
+		}
+		plans = append(plans, node)
+	}
+
+	return plans, nil
+}

pkg/telemetrylogs/json_access_pb_test.go

@@ -0,0 +1,903 @@
+package telemetrylogs
+
+import (
+	"context"
+	"testing"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/stretchr/testify/require"
+)
+
+// ============================================================================
+// Helper Functions for Test Data Creation
+// ============================================================================
+
+// makeKey creates a TelemetryFieldKey for testing
+func makeKey(name string, dataType telemetrytypes.JSONDataType, materialized bool) *telemetrytypes.TelemetryFieldKey {
+	return &telemetrytypes.TelemetryFieldKey{
+		Name:         name,
+		JSONDataType: &dataType,
+		Materialized: materialized,
+	}
+}
+
+// inferDataTypeFromValue infers JSONDataType from a Go value
+func inferDataTypeFromValue(value any) telemetrytypes.JSONDataType {
+	switch v := value.(type) {
+	case string:
+		return telemetrytypes.String
+	case int64:
+		return telemetrytypes.Int64
+	case int:
+		return telemetrytypes.Int64
+	case float64:
+		return telemetrytypes.Float64
+	case float32:
+		return telemetrytypes.Float64
+	case bool:
+		return telemetrytypes.Bool
+	case []any:
+		if len(v) == 0 {
+			return telemetrytypes.Dynamic
+		}
+		return inferDataTypeFromValue(v[0])
+	case nil:
+		return telemetrytypes.String
+	default:
+		return telemetrytypes.String
+	}
+}
+
+// makeGetTypes creates a getTypes function from a map of path -> types
+func makeGetTypes(typesMap map[string][]telemetrytypes.JSONDataType) func(ctx context.Context, path string) ([]telemetrytypes.JSONDataType, error) {
+	return func(_ context.Context, path string) ([]telemetrytypes.JSONDataType, error) {
+		return typesMap[path], nil
+	}
+}
+
+// ============================================================================
+// Helper Functions for Node Validation
+// ============================================================================
+
+// findTerminalNode finds the terminal node in a plan tree
+func findTerminalNode(node *telemetrytypes.JSONAccessNode) *telemetrytypes.JSONAccessNode {
+	if node == nil {
+		return nil
+	}
+	if node.IsTerminal {
+		return node
+	}
+	if node.Branches[telemetrytypes.BranchJSON] != nil {
+		return findTerminalNode(node.Branches[telemetrytypes.BranchJSON])
+	}
+	if node.Branches[telemetrytypes.BranchDynamic] != nil {
+		return findTerminalNode(node.Branches[telemetrytypes.BranchDynamic])
+	}
+	return nil
+}
+
+// validateTerminalNode validates a terminal node has expected properties
+func validateTerminalNode(t *testing.T, node *telemetrytypes.JSONAccessNode, expectedName string, expectedElemType telemetrytypes.JSONDataType) {
+	require.NotNil(t, node, "terminal node should not be nil")
+	require.True(t, node.IsTerminal, "node should be terminal")
+	require.Equal(t, expectedName, node.Name, "node name mismatch")
+	require.NotNil(t, node.TerminalConfig, "terminal config should not be nil")
+	require.Equal(t, expectedElemType, node.TerminalConfig.ElemType, "elem type mismatch")
+}
+
+// validateNodeStructure validates basic node structure
+func validateNodeStructure(t *testing.T, node *telemetrytypes.JSONAccessNode, expectedName string, isTerminal bool) {
+	require.NotNil(t, node, "node should not be nil")
+	require.Equal(t, expectedName, node.Name, "node name mismatch")
+	require.Equal(t, isTerminal, node.IsTerminal, "isTerminal mismatch")
+}
+
+// validateRootNode validates root node structure
+func validateRootNode(t *testing.T, plan *telemetrytypes.JSONAccessNode, expectedColumn string, expectedMaxPaths int) {
+	require.NotNil(t, plan, "plan should not be nil")
+	require.NotNil(t, plan.Parent, "root parent should not be nil")
+	require.Equal(t, expectedColumn, plan.Parent.Name, "root column name mismatch")
+	require.Equal(t, expectedMaxPaths, plan.Parent.MaxDynamicPaths, "root MaxDynamicPaths mismatch")
+}
+
+// validateBranchExists validates that a branch exists and optionally checks its properties
+func validateBranchExists(t *testing.T, node *telemetrytypes.JSONAccessNode, branchType telemetrytypes.JSONAccessBranchType, expectedMaxTypes *int, expectedMaxPaths *int) {
+	require.NotNil(t, node, "node should not be nil")
+	branch := node.Branches[branchType]
+	require.NotNil(t, branch, "branch %v should exist", branchType)
+	if expectedMaxTypes != nil {
+		require.Equal(t, *expectedMaxTypes, branch.MaxDynamicTypes, "MaxDynamicTypes mismatch for branch %v", branchType)
+	}
+	if expectedMaxPaths != nil {
+		require.Equal(t, *expectedMaxPaths, branch.MaxDynamicPaths, "MaxDynamicPaths mismatch for branch %v", branchType)
+	}
+}
+
+// validateMaxDynamicTypesProgression validates MaxDynamicTypes progression through nested levels
+func validateMaxDynamicTypesProgression(t *testing.T, node *telemetrytypes.JSONAccessNode, expectedValues []int) {
+	current := node
+	for i, expected := range expectedValues {
+		if current == nil {
+			t.Fatalf("node is nil at level %d", i)
+		}
+		require.Equal(t, expected, current.MaxDynamicTypes, "MaxDynamicTypes mismatch at level %d (node: %s)", i, current.Name)
+		if current.Branches[telemetrytypes.BranchJSON] != nil {
+			current = current.Branches[telemetrytypes.BranchJSON]
+		} else {
+			break
+		}
+	}
+}
+
+// ============================================================================
+// Test Cases for Node Methods
+// ============================================================================
+
+func TestNode_Alias(t *testing.T) {
+	tests := []struct {
+		name     string
+		node     *telemetrytypes.JSONAccessNode
+		expected string
+	}{
+		{
+			name:     "Root node returns name as-is",
+			node:     telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			expected: LogsV2BodyJSONColumn,
+		},
+		{
+			name: "Node without parent returns backticked name",
+			node: &telemetrytypes.JSONAccessNode{
+				Name:   "user",
+				Parent: nil,
+			},
+			expected: "`user`",
+		},
+		{
+			name: "Node with root parent uses dot separator",
+			node: &telemetrytypes.JSONAccessNode{
+				Name:   "age",
+				Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			},
+			expected: "`" + LogsV2BodyJSONColumn + ".age`",
+		},
+		{
+			name: "Node with non-root parent uses array separator",
+			node: &telemetrytypes.JSONAccessNode{
+				Name: "name",
+				Parent: &telemetrytypes.JSONAccessNode{
+					Name:   "education",
+					Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+				},
+			},
+			expected: "`" + LogsV2BodyJSONColumn + ".education[].name`",
+		},
+		{
+			name: "Nested array path with multiple levels",
+			node: &telemetrytypes.JSONAccessNode{
+				Name: "type",
+				Parent: &telemetrytypes.JSONAccessNode{
+					Name: "awards",
+					Parent: &telemetrytypes.JSONAccessNode{
+						Name:   "education",
+						Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+					},
+				},
+			},
+			expected: "`" + LogsV2BodyJSONColumn + ".education[].awards[].type`",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.node.Alias()
+			require.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestNode_FieldPath(t *testing.T) {
+	tests := []struct {
+		name     string
+		node     *telemetrytypes.JSONAccessNode
+		expected string
+	}{
+		{
+			name: "Simple field path from root",
+			node: &telemetrytypes.JSONAccessNode{
+				Name:   "user",
+				Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			},
+			// FieldPath() always wraps the field name in backticks
+			expected: LogsV2BodyJSONColumn + ".`user`",
+		},
+		{
+			name: "Field path with backtick-required key",
+			node: &telemetrytypes.JSONAccessNode{
+				Name:   "user-name", // requires backtick
+				Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			},
+			expected: LogsV2BodyJSONColumn + ".`user-name`",
+		},
+		{
+			name: "Nested field path",
+			node: &telemetrytypes.JSONAccessNode{
+				Name: "age",
+				Parent: &telemetrytypes.JSONAccessNode{
+					Name:   "user",
+					Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+				},
+			},
+			// FieldPath() always wraps the field name in backticks
+			expected: "`" + LogsV2BodyJSONColumn + ".user`.`age`",
+		},
+		{
+			name: "Array element field path",
+			node: &telemetrytypes.JSONAccessNode{
+				Name: "name",
+				Parent: &telemetrytypes.JSONAccessNode{
+					Name:   "education",
+					Parent: telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+				},
+			},
+			// FieldPath() always wraps the field name in backticks
+			expected: "`" + LogsV2BodyJSONColumn + ".education`.`name`",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.node.FieldPath()
+			require.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+// ============================================================================
+// Test Cases for buildPlan
+// ============================================================================
+
+func TestPlanBuilder_buildPlan(t *testing.T) {
+	tests := []struct {
+		name          string
+		parts         []string
+		key           *telemetrytypes.TelemetryFieldKey
+		getTypes      func(ctx context.Context, path string) ([]telemetrytypes.JSONDataType, error)
+		isDynArrChild bool
+		parent        *telemetrytypes.JSONAccessNode
+		validate      func(t *testing.T, node *telemetrytypes.JSONAccessNode)
+	}{
+		{
+			name:  "Simple path with single part",
+			parts: []string{"user"},
+			key:   makeKey("user", telemetrytypes.String, false),
+			getTypes: makeGetTypes(map[string][]telemetrytypes.JSONDataType{
+				"user": {telemetrytypes.String},
+			}),
+			isDynArrChild: false,
+			parent:        telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			validate: func(t *testing.T, node *telemetrytypes.JSONAccessNode) {
+				validateTerminalNode(t, node, "user", telemetrytypes.String)
+			},
+		},
+		{
+			name:  "Path with array - JSON branch",
+			parts: []string{"education", "name"},
+			key:   makeKey("education[].name", telemetrytypes.String, false),
+			getTypes: makeGetTypes(map[string][]telemetrytypes.JSONDataType{
+				"education":        {telemetrytypes.ArrayJSON},
+				"education[].name": {telemetrytypes.String},
+			}),
+			isDynArrChild: false,
+			parent:        telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			validate: func(t *testing.T, node *telemetrytypes.JSONAccessNode) {
+				validateNodeStructure(t, node, "education", false)
+				require.Equal(t, 16, node.MaxDynamicTypes) // 32/2
+				require.NotNil(t, node.Branches[telemetrytypes.BranchJSON])
+				child := node.Branches[telemetrytypes.BranchJSON]
+				require.True(t, child.IsTerminal)
+				require.Equal(t, 8, child.MaxDynamicTypes) // 16/2
+			},
+		},
+		{
+			name:  "Path with array - Dynamic branch",
+			parts: []string{"education", "name"},
+			key:   makeKey("education[].name", telemetrytypes.String, false),
+			getTypes: makeGetTypes(map[string][]telemetrytypes.JSONDataType{
+				"education":        {telemetrytypes.ArrayDynamic},
+				"education[].name": {telemetrytypes.String},
+			}),
+			isDynArrChild: false,
+			parent:        telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			validate: func(t *testing.T, node *telemetrytypes.JSONAccessNode) {
+				validateNodeStructure(t, node, "education", false)
+				expectedMaxTypes := 16
+				expectedMaxPaths := 256
+				validateBranchExists(t, node, telemetrytypes.BranchDynamic, &expectedMaxTypes, &expectedMaxPaths)
+			},
+		},
+		{
+			name:  "Path with both JSON and Dynamic branches",
+			parts: []string{"education", "name"},
+			key:   makeKey("education[].name", telemetrytypes.String, false),
+			getTypes: makeGetTypes(map[string][]telemetrytypes.JSONDataType{
+				"education":        {telemetrytypes.ArrayJSON, telemetrytypes.ArrayDynamic},
+				"education[].name": {telemetrytypes.String},
+			}),
+			isDynArrChild: false,
+			parent:        telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			validate: func(t *testing.T, node *telemetrytypes.JSONAccessNode) {
+				validateNodeStructure(t, node, "education", false)
+				require.NotNil(t, node.Branches[telemetrytypes.BranchJSON])
+				require.NotNil(t, node.Branches[telemetrytypes.BranchDynamic])
+			},
+		},
+		{
+			name:  "Nested array path progression",
+			parts: []string{"education", "awards", "type"},
+			key:   makeKey("education[].awards[].type", telemetrytypes.String, false),
+			getTypes: makeGetTypes(map[string][]telemetrytypes.JSONDataType{
+				"education":                 {telemetrytypes.ArrayJSON},
+				"education[].awards":        {telemetrytypes.ArrayJSON},
+				"education[].awards[].type": {telemetrytypes.String},
+			}),
+			isDynArrChild: false,
+			parent:        telemetrytypes.NewRootJSONAccessNode(LogsV2BodyJSONColumn, 32, 0),
+			validate: func(t *testing.T, node *telemetrytypes.JSONAccessNode) {
+				validateNodeStructure(t, node, "education", false)
+				require.Equal(t, 16, node.MaxDynamicTypes) // 32/2
+				child := node.Branches[telemetrytypes.BranchJSON]
+				require.Equal(t, 8, child.MaxDynamicTypes) // 16/2
+				grandchild := child.Branches[telemetrytypes.BranchJSON]
+				require.Equal(t, 4, grandchild.MaxDynamicTypes) // 8/2
+				require.True(t, grandchild.IsTerminal)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pb := &JSONAccessPlanBuilder{
+				key:      tt.key,
+				parts:    tt.parts,
+				getTypes: tt.getTypes,
+			}
+			result, err := pb.buildPlan(context.Background(), 0, tt.parent, tt.isDynArrChild)
+			require.NoError(t, err)
+			tt.validate(t, result)
+		})
+	}
+}
+
+// ============================================================================
+// Test Cases for PlanJSON
+// ============================================================================
+
+func TestPlanJSON_BasicStructure(t *testing.T) {
+	_, getTypes := testTypeSet()
+
+	tests := []struct {
+		name      string
+		key       *telemetrytypes.TelemetryFieldKey
+		expectErr bool
+		validate  func(t *testing.T, plans []*telemetrytypes.JSONAccessNode)
+	}{
+		{
+			name: "Simple path not promoted",
+			key:  makeKey("user.name", telemetrytypes.String, false),
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				require.Len(t, plans, 1)
+				validateRootNode(t, plans[0], LogsV2BodyJSONColumn, 0)
+			},
+		},
+		{
+			name: "Simple path promoted",
+			key:  makeKey("user.name", telemetrytypes.String, true),
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				require.Len(t, plans, 2)
+				validateRootNode(t, plans[0], LogsV2BodyJSONColumn, 0)
+				validateRootNode(t, plans[1], LogsV2BodyPromotedColumn, 1024)
+				require.Equal(t, plans[0].Name, plans[1].Name)
+			},
+		},
+		{
+			name:      "Empty path returns error",
+			key:       makeKey("", telemetrytypes.String, false),
+			expectErr: true,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				require.Nil(t, plans)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			plans, err := PlanJSON(context.Background(), tt.key, qbtypes.FilterOperatorEqual, "John", getTypes)
+			if tt.expectErr {
+				require.Error(t, err)
+				tt.validate(t, plans)
+				return
+			}
+			require.NoError(t, err)
+			tt.validate(t, plans)
+		})
+	}
+}
+
+func TestPlanJSON_Operators(t *testing.T) {
+	_, getTypes := testTypeSet()
+
+	tests := []struct {
+		name     string
+		path     string
+		operator qbtypes.FilterOperator
+		value    any
+		validate func(t *testing.T, terminal *telemetrytypes.JSONAccessNode)
+	}{
+		{
+			name:     "Equal operator with string",
+			path:     "user.name",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    "John",
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				// Path "user.name" is not split by ".", so terminal node name is "user.name"
+				validateTerminalNode(t, terminal, "user.name", telemetrytypes.String)
+			},
+		},
+		{
+			name:     "NotEqual operator with int64",
+			path:     "user.age",
+			operator: qbtypes.FilterOperatorNotEqual,
+			value:    int64(30),
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				// Path "user.age" is not split by ".", so terminal node name is "user.age"
+				validateTerminalNode(t, terminal, "user.age", telemetrytypes.Int64)
+			},
+		},
+		{
+			name:     "Contains operator with string",
+			path:     "education[].name",
+			operator: qbtypes.FilterOperatorContains,
+			value:    "IIT",
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				validateTerminalNode(t, terminal, "name", telemetrytypes.String)
+			},
+		},
+		{
+			name:     "Contains operator with array parameter",
+			path:     "education[].parameters",
+			operator: qbtypes.FilterOperatorContains,
+			value:    1.65,
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				// Terminal config uses key's JSONDataType (inferred from value), not available types
+				validateTerminalNode(t, terminal, "parameters", telemetrytypes.Float64)
+			},
+		},
+		{
+			name:     "In operator with array value",
+			path:     "user.name",
+			operator: qbtypes.FilterOperatorIn,
+			value:    []any{"John", "Jane", "Bob"},
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				// Path "user.name" is not split by ".", so terminal node name is "user.name"
+				validateTerminalNode(t, terminal, "user.name", telemetrytypes.String)
+			},
+		},
+		{
+			name:     "Exists operator with nil",
+			path:     "user.age",
+			operator: qbtypes.FilterOperatorExists,
+			value:    nil,
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				// Path "user.age" is not split by ".", so terminal node name is "user.age"
+				// Value is nil, so type is inferred as String, but test expects Int64
+				// This test should use Int64 type when creating the key
+				require.NotNil(t, terminal)
+				require.NotNil(t, terminal.TerminalConfig)
+			},
+		},
+		{
+			name:     "Like operator",
+			path:     "user.name",
+			operator: qbtypes.FilterOperatorLike,
+			value:    "John%",
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				require.NotNil(t, terminal)
+				require.NotNil(t, terminal.TerminalConfig)
+			},
+		},
+		{
+			name:     "GreaterThan operator",
+			path:     "user.age",
+			operator: qbtypes.FilterOperatorGreaterThan,
+			value:    int64(18),
+			validate: func(t *testing.T, terminal *telemetrytypes.JSONAccessNode) {
+				require.NotNil(t, terminal)
+				require.NotNil(t, terminal.TerminalConfig)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// For Exists operator with nil value on user.age, use Int64 type
+			dataType := inferDataTypeFromValue(tt.value)
+			if tt.path == "user.age" && tt.operator == qbtypes.FilterOperatorExists {
+				dataType = telemetrytypes.Int64
+			}
+			key := makeKey(tt.path, dataType, false)
+			plans, err := PlanJSON(context.Background(), key, tt.operator, tt.value, getTypes)
+			require.NoError(t, err)
+			require.NotNil(t, plans)
+			require.Len(t, plans, 1)
+			terminal := findTerminalNode(plans[0])
+			tt.validate(t, terminal)
+		})
+	}
+}
+
+func TestPlanJSON_ArrayPaths(t *testing.T) {
+	_, getTypes := testTypeSet()
+
+	tests := []struct {
+		name     string
+		path     string
+		operator qbtypes.FilterOperator
+		validate func(t *testing.T, plans []*telemetrytypes.JSONAccessNode)
+	}{
+		{
+			name:     "Single array level - JSON branch only",
+			path:     "education[].name",
+			operator: qbtypes.FilterOperatorEqual,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				node := plans[0]
+				validateNodeStructure(t, node, "education", false)
+				require.NotNil(t, node.Branches[telemetrytypes.BranchJSON])
+				require.Nil(t, node.Branches[telemetrytypes.BranchDynamic])
+				child := node.Branches[telemetrytypes.BranchJSON]
+				validateTerminalNode(t, child, "name", telemetrytypes.String)
+				require.Equal(t, 8, child.MaxDynamicTypes) // 16/2
+			},
+		},
+		{
+			name:     "Single array level - both JSON and Dynamic branches",
+			path:     "education[].awards[].type",
+			operator: qbtypes.FilterOperatorEqual,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				node := plans[0]
+				validateNodeStructure(t, node, "education", false)
+				require.NotNil(t, node.Branches[telemetrytypes.BranchJSON])
+				child := node.Branches[telemetrytypes.BranchJSON]
+				require.Equal(t, "awards", child.Name)
+				require.NotNil(t, child.Branches[telemetrytypes.BranchJSON])
+				require.NotNil(t, child.Branches[telemetrytypes.BranchDynamic])
+				terminalJSON := findTerminalNode(child.Branches[telemetrytypes.BranchJSON])
+				terminalDyn := findTerminalNode(child.Branches[telemetrytypes.BranchDynamic])
+				require.Equal(t, 4, terminalJSON.MaxDynamicTypes)
+				require.Equal(t, 16, terminalDyn.MaxDynamicTypes)  // Reset for Dynamic
+				require.Equal(t, 256, terminalDyn.MaxDynamicPaths) // Reset for Dynamic
+			},
+		},
+		{
+			name:     "Deeply nested array path",
+			path:     "interests[].entities[].reviews[].entries[].metadata[].positions[].name",
+			operator: qbtypes.FilterOperatorEqual,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				node := plans[0]
+				expectedTypes := []int{16, 8, 4, 2, 1, 0}
+				validateMaxDynamicTypesProgression(t, node, expectedTypes)
+				terminal := findTerminalNode(node)
+				require.True(t, terminal.IsTerminal)
+			},
+		},
+		{
+			name:     "ArrayAnyIndex replacement [*] to []",
+			path:     "education[*].name",
+			operator: qbtypes.FilterOperatorEqual,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				node := plans[0]
+				validateNodeStructure(t, node, "education", false)
+				require.NotNil(t, node.Branches[telemetrytypes.BranchJSON])
+				terminal := findTerminalNode(node.Branches[telemetrytypes.BranchJSON])
+				require.NotNil(t, terminal)
+				require.Equal(t, "name", terminal.Name)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			key := makeKey(tt.path, telemetrytypes.String, false)
+			plans, err := PlanJSON(context.Background(), key, tt.operator, "John", getTypes)
+			require.NoError(t, err)
+			require.NotNil(t, plans)
+			require.Len(t, plans, 1)
+			tt.validate(t, plans)
+		})
+	}
+}
+
+func TestPlanJSON_ArrayMembership(t *testing.T) {
+	_, getTypes := testTypeSet()
+
+	tests := []struct {
+		name             string
+		path             string
+		operator         qbtypes.FilterOperator
+		value            any
+		expectedElemType telemetrytypes.JSONDataType
+	}{
+		{
+			name:     "Contains with ArrayFloat64",
+			path:     "education[].parameters",
+			operator: qbtypes.FilterOperatorContains,
+			value:    1.65,
+			// Terminal config uses key's JSONDataType (inferred from value), not available types
+			expectedElemType: telemetrytypes.Float64,
+		},
+		{
+			name:     "Contains with ArrayString",
+			path:     "education[].parameters",
+			operator: qbtypes.FilterOperatorContains,
+			value:    "passed",
+			// Terminal config uses key's JSONDataType (inferred from value), not available types
+			expectedElemType: telemetrytypes.String,
+		},
+		{
+			name:     "Contains with ArrayInt64",
+			path:     "interests[].entities[].reviews[].entries[].metadata[].positions[].ratings",
+			operator: qbtypes.FilterOperatorContains,
+			value:    int64(4),
+			// Terminal config uses key's JSONDataType (inferred from value), not available types
+			expectedElemType: telemetrytypes.Int64,
+		},
+		{
+			name:             "Contains with scalar only",
+			path:             "education[].name",
+			operator:         qbtypes.FilterOperatorContains,
+			value:            "IIT",
+			expectedElemType: telemetrytypes.String,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			key := makeKey(tt.path, inferDataTypeFromValue(tt.value), false)
+			plans, err := PlanJSON(context.Background(), key, tt.operator, tt.value, getTypes)
+			require.NoError(t, err)
+			require.NotNil(t, plans)
+			require.Len(t, plans, 1)
+			terminal := findTerminalNode(plans[0])
+			require.NotNil(t, terminal)
+			require.NotNil(t, terminal.TerminalConfig)
+			require.Equal(t, tt.expectedElemType, terminal.TerminalConfig.ElemType)
+		})
+	}
+}
+
+func TestPlanJSON_PromotedVsNonPromoted(t *testing.T) {
+	_, getTypes := testTypeSet()
+	path := "education[].awards[].type"
+	value := "sports"
+
+	t.Run("Non-promoted plan", func(t *testing.T) {
+		key := makeKey(path, inferDataTypeFromValue(value), false)
+		plans, err := PlanJSON(context.Background(), key, qbtypes.FilterOperatorEqual, value, getTypes)
+		require.NoError(t, err)
+		require.Len(t, plans, 1)
+		validateRootNode(t, plans[0], LogsV2BodyJSONColumn, 0)
+		require.Equal(t, 0, plans[0].MaxDynamicPaths)
+	})
+
+	t.Run("Promoted plan", func(t *testing.T) {
+		key := makeKey(path, inferDataTypeFromValue(value), true)
+		plans, err := PlanJSON(context.Background(), key, qbtypes.FilterOperatorEqual, value, getTypes)
+		require.NoError(t, err)
+		require.Len(t, plans, 2)
+		validateRootNode(t, plans[0], LogsV2BodyJSONColumn, 0)
+		validateRootNode(t, plans[1], LogsV2BodyPromotedColumn, 1024)
+
+		terminal1 := findTerminalNode(plans[0])
+		terminal2 := findTerminalNode(plans[1])
+		require.NotNil(t, terminal1)
+		require.NotNil(t, terminal2)
+		require.Equal(t, terminal1.Name, terminal2.Name)
+
+		// Check MaxDynamicPaths progression
+		node1 := plans[0]
+		node2 := plans[1]
+		require.Equal(t, 256, node2.MaxDynamicPaths, "Promoted education node should have 256")
+		require.Equal(t, 0, node1.MaxDynamicPaths, "Non-promoted education node should have 0")
+
+		if node1.Branches[telemetrytypes.BranchJSON] != nil && node2.Branches[telemetrytypes.BranchJSON] != nil {
+			child1 := node1.Branches[telemetrytypes.BranchJSON]
+			child2 := node2.Branches[telemetrytypes.BranchJSON]
+			require.Equal(t, 64, child2.MaxDynamicPaths, "Promoted awards node should have 64")
+			require.Equal(t, 0, child1.MaxDynamicPaths, "Non-promoted awards node should have 0")
+		}
+	})
+}
+
+func TestPlanJSON_EdgeCases(t *testing.T) {
+	_, getTypes := testTypeSet()
+
+	tests := []struct {
+		name     string
+		path     string
+		operator qbtypes.FilterOperator
+		value    any
+		validate func(t *testing.T, plans []*telemetrytypes.JSONAccessNode)
+	}{
+		{
+			name:     "Path with no available types",
+			path:     "unknown.path",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    "test",
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				require.NotNil(t, plans)
+				require.Len(t, plans, 1)
+				terminal := findTerminalNode(plans[0])
+				if terminal != nil {
+					require.NotNil(t, terminal.TerminalConfig)
+					require.Equal(t, telemetrytypes.String, terminal.TerminalConfig.ElemType)
+				}
+			},
+		},
+		{
+			name:     "Very deep nesting - validates progression doesn't go negative",
+			path:     "interests[].entities[].reviews[].entries[].metadata[].positions[].name",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    "Engineer",
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				node := plans[0]
+				current := node
+				for !current.IsTerminal && current.Branches[telemetrytypes.BranchJSON] != nil {
+					require.GreaterOrEqual(t, current.MaxDynamicTypes, 0,
+						"MaxDynamicTypes should not be negative at node %s", current.Name)
+					require.GreaterOrEqual(t, current.MaxDynamicPaths, 0,
+						"MaxDynamicPaths should not be negative at node %s", current.Name)
+					current = current.Branches[telemetrytypes.BranchJSON]
+				}
+			},
+		},
+		{
+			name:     "Path with mixed scalar and array types",
+			path:     "education[].type",
+			operator: qbtypes.FilterOperatorEqual,
+			value:    "high_school",
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				terminal := findTerminalNode(plans[0])
+				require.NotNil(t, terminal)
+				require.Contains(t, terminal.AvailableTypes, telemetrytypes.String)
+				require.Contains(t, terminal.AvailableTypes, telemetrytypes.Int64)
+				require.Equal(t, telemetrytypes.String, terminal.TerminalConfig.ElemType)
+			},
+		},
+		{
+			name:     "Exists with only array types available",
+			path:     "education",
+			operator: qbtypes.FilterOperatorExists,
+			value:    nil,
+			validate: func(t *testing.T, plans []*telemetrytypes.JSONAccessNode) {
+				terminal := findTerminalNode(plans[0])
+				require.NotNil(t, terminal)
+				require.NotNil(t, terminal.TerminalConfig)
+				// When path is an array and value is nil, key should use ArrayJSON type
+				require.Equal(t, telemetrytypes.ArrayJSON, terminal.TerminalConfig.ElemType)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// For "education" path with Exists operator, use ArrayJSON type
+			dataType := inferDataTypeFromValue(tt.value)
+			if tt.path == "education" && tt.operator == qbtypes.FilterOperatorExists {
+				dataType = telemetrytypes.ArrayJSON
+			}
+			key := makeKey(tt.path, dataType, false)
+			plans, err := PlanJSON(context.Background(), key, tt.operator, tt.value, getTypes)
+			require.NoError(t, err)
+			tt.validate(t, plans)
+		})
+	}
+}
+
+func TestPlanJSON_TreeStructure(t *testing.T) {
+	_, getTypes := testTypeSet()
+	path := "education[].awards[].participated[].team[].branch"
+	key := makeKey(path, telemetrytypes.String, false)
+	plans, err := PlanJSON(context.Background(), key, qbtypes.FilterOperatorEqual, "John", getTypes)
+	require.NoError(t, err)
+	require.Len(t, plans, 1)
+
+	node := plans[0]
+	var validateNode func(*telemetrytypes.JSONAccessNode)
+	validateNode = func(n *telemetrytypes.JSONAccessNode) {
+		if n == nil {
+			return
+		}
+		if n.Parent != nil {
+			require.NotNil(t, n.Parent, "Node %s should have parent", n.Name)
+			if !n.IsTerminal && n.Parent != nil {
+				require.False(t, n.Parent.IsTerminal,
+					"Non-terminal node %s should have non-terminal parent", n.Name)
+			}
+		}
+		if n.Branches[telemetrytypes.BranchJSON] != nil {
+			validateNode(n.Branches[telemetrytypes.BranchJSON])
+		}
+		if n.Branches[telemetrytypes.BranchDynamic] != nil {
+			validateNode(n.Branches[telemetrytypes.BranchDynamic])
+		}
+	}
+	validateNode(node)
+}
+
+// ============================================================================
+// Test Data Setup
+// ============================================================================
+
+// testTypeSet returns a map of path->types and a getTypes function for testing
+// This represents the type information available in the test JSON structure
+func testTypeSet() (map[string][]telemetrytypes.JSONDataType, func(ctx context.Context, path string) ([]telemetrytypes.JSONDataType, error)) {
+	types := map[string][]telemetrytypes.JSONDataType{
+		"user.name":                                           {telemetrytypes.String},
+		"user.age":                                            {telemetrytypes.Int64, telemetrytypes.String},
+		"user.height":                                         {telemetrytypes.Float64},
+		"education":                                           {telemetrytypes.ArrayJSON},
+		"education[].name":                                    {telemetrytypes.String},
+		"education[].type":                                    {telemetrytypes.String, telemetrytypes.Int64},
+		"education[].internal_type":                           {telemetrytypes.String},
+		"education[].metadata.location":                       {telemetrytypes.String},
+		"education[].parameters":                              {telemetrytypes.ArrayFloat64, telemetrytypes.ArrayDynamic},
+		"education[].duration":                                {telemetrytypes.String},
+		"education[].mode":                                    {telemetrytypes.String},
+		"education[].year":                                    {telemetrytypes.Int64},
+		"education[].field":                                   {telemetrytypes.String},
+		"education[].awards":                                  {telemetrytypes.ArrayDynamic, telemetrytypes.ArrayJSON},
+		"education[].awards[].name":                           {telemetrytypes.String},
+		"education[].awards[].rank":                           {telemetrytypes.Int64},
+		"education[].awards[].medal":                          {telemetrytypes.String},
+		"education[].awards[].type":                           {telemetrytypes.String},
+		"education[].awards[].semester":                       {telemetrytypes.Int64},
+		"education[].awards[].participated":                   {telemetrytypes.ArrayDynamic, telemetrytypes.ArrayJSON},
+		"education[].awards[].participated[].type":            {telemetrytypes.String},
+		"education[].awards[].participated[].field":           {telemetrytypes.String},
+		"education[].awards[].participated[].project_type":    {telemetrytypes.String},
+		"education[].awards[].participated[].project_name":    {telemetrytypes.String},
+		"education[].awards[].participated[].race_type":       {telemetrytypes.String},
+		"education[].awards[].participated[].team_based":      {telemetrytypes.Bool},
+		"education[].awards[].participated[].team_name":       {telemetrytypes.String},
+		"education[].awards[].participated[].team":            {telemetrytypes.ArrayJSON},
+		"education[].awards[].participated[].team[].name":     {telemetrytypes.String},
+		"education[].awards[].participated[].team[].branch":   {telemetrytypes.String},
+		"education[].awards[].participated[].team[].semester": {telemetrytypes.Int64},
+		"interests":                                                                  {telemetrytypes.ArrayJSON},
+		"interests[].type":                                                           {telemetrytypes.String},
+		"interests[].entities":                                                       {telemetrytypes.ArrayJSON},
+		"interests[].entities.application_date":                                      {telemetrytypes.String},
+		"interests[].entities[].reviews":                                             {telemetrytypes.ArrayJSON},
+		"interests[].entities[].reviews[].given_by":                                  {telemetrytypes.String},
+		"interests[].entities[].reviews[].remarks":                                   {telemetrytypes.String},
+		"interests[].entities[].reviews[].weight":                                    {telemetrytypes.Float64},
+		"interests[].entities[].reviews[].passed":                                    {telemetrytypes.Bool},
+		"interests[].entities[].reviews[].type":                                      {telemetrytypes.String},
+		"interests[].entities[].reviews[].analysis_type":                             {telemetrytypes.Int64},
+		"interests[].entities[].reviews[].entries":                                   {telemetrytypes.ArrayJSON},
+		"interests[].entities[].reviews[].entries[].subject":                         {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].status":                          {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].metadata":                        {telemetrytypes.ArrayJSON},
+		"interests[].entities[].reviews[].entries[].metadata[].company":              {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].metadata[].experience":           {telemetrytypes.Int64},
+		"interests[].entities[].reviews[].entries[].metadata[].unit":                 {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].metadata[].positions":            {telemetrytypes.ArrayJSON},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].name":     {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].duration": {telemetrytypes.Int64, telemetrytypes.Float64},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].unit":     {telemetrytypes.String},
+		"interests[].entities[].reviews[].entries[].metadata[].positions[].ratings":  {telemetrytypes.ArrayInt64, telemetrytypes.ArrayString},
+		"message": {telemetrytypes.String},
+	}
+
+	return types, makeGetTypes(types)
+}

pkg/telemetrylogs/statement_builder.go

@@ -589,10 +589,9 @@ func (b *logQueryStatementBuilder) addFilterCondition(
 			FieldKeys:          keys,
 			SkipResourceFilter: true,
 			FullTextColumn:     b.fullTextColumn,
-			JsonBodyPrefix:     b.jsonBodyPrefix,
 			JsonKeyToKey:       b.jsonKeyToKey,
 			Variables:          variables,
-        }, start, end)
+		}, start, end)
 
 		if err != nil {
 			return nil, err

pkg/telemetrymetadata/body_json_metadata.go

@@ -0,0 +1,440 @@
+package telemetrymetadata
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"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"
+	"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"
+)
+
+var (
+	defaultPathLimit = 100 // Default limit to prevent full table scans
+
+	CodeUnknownJSONDataType     = errors.MustNewCode("unknown_json_data_type")
+	CodeFailLoadPromotedPaths   = errors.MustNewCode("fail_load_promoted_paths")
+	CodeFailCheckPathPromoted   = errors.MustNewCode("fail_check_path_promoted")
+	CodeFailIterateBodyJSONKeys = errors.MustNewCode("fail_iterate_body_json_keys")
+	CodeFailExtractBodyJSONKeys = errors.MustNewCode("fail_extract_body_json_keys")
+	CodeFailLoadLogsJSONIndexes = errors.MustNewCode("fail_load_logs_json_indexes")
+	CodeFailListJSONValues      = errors.MustNewCode("fail_list_json_values")
+	CodeFailScanJSONValue       = errors.MustNewCode("fail_scan_json_value")
+	CodeFailScanVariant         = errors.MustNewCode("fail_scan_variant")
+	CodeFailBuildJSONPathsQuery = errors.MustNewCode("fail_build_json_paths_query")
+)
+
+// GetBodyJSONPaths extracts body JSON paths from the path_types table
+// This function can be used by both JSONQueryBuilder and metadata extraction
+// uniquePathLimit: 0 for no limit, >0 for maximum number of unique paths to return
+//   - For startup load: set to 10000 to get top 10k unique paths
+//   - For lookup: set to 0 (no limit needed for single path)
+//   - For metadata API: set to desired pagination limit
+//
+// searchOperator: LIKE for pattern matching, EQUAL for exact match
+// Returns: (paths, error)
+func getBodyJSONPaths(ctx context.Context, telemetryStore telemetrystore.TelemetryStore,
+	fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
+
+	query, args, limit, err := buildGetBodyJSONPathsQuery(fieldKeySelectors)
+	if err != nil {
+		return nil, false, err
+	}
+
+	rows, err := telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, false, errors.WrapInternalf(err, CodeFailExtractBodyJSONKeys, "failed to extract body JSON keys")
+	}
+	defer rows.Close()
+
+	paths := []*telemetrytypes.TelemetryFieldKey{}
+	rowCount := 0
+	for rows.Next() {
+		var path string
+		var typesArray []string // ClickHouse returns array as []string
+		var lastSeen uint64
+
+		err = rows.Scan(&path, &typesArray, &lastSeen)
+		if err != nil {
+			return nil, false, errors.WrapInternalf(err, CodeFailExtractBodyJSONKeys, "failed to scan body JSON key row")
+		}
+
+		promoted, err := IsPathPromoted(ctx, telemetryStore.ClickhouseDB(), path)
+		if err != nil {
+			return nil, false, err
+		}
+
+		indexes, err := getJSONPathIndexes(ctx, telemetryStore, path)
+		if err != nil {
+			return nil, false, err
+		}
+
+		for _, typ := range typesArray {
+			mapping, found := telemetrytypes.MappingStringToJSONDataType[typ]
+			if !found {
+				return nil, false, errors.NewInternalf(CodeUnknownJSONDataType, "failed to map type string to JSON data type: %s", typ)
+			}
+			paths = append(paths, &telemetrytypes.TelemetryFieldKey{
+				Name:          path,
+				Signal:        telemetrytypes.SignalLogs,
+				FieldContext:  telemetrytypes.FieldContextBody,
+				FieldDataType: telemetrytypes.MappingJSONDataTypeToFieldDataType[mapping],
+				JSONDataType:  &mapping,
+				Indexes:       indexes,
+				Materialized:  promoted,
+			})
+		}
+
+		rowCount++
+	}
+
+	if rows.Err() != nil {
+		return nil, false, errors.WrapInternalf(rows.Err(), CodeFailIterateBodyJSONKeys, "error iterating body JSON keys")
+	}
+
+	return paths, rowCount <= limit, nil
+}
+
+func buildGetBodyJSONPathsQuery(fieldKeySelectors []*telemetrytypes.FieldKeySelector) (string, []any, int, error) {
+	if len(fieldKeySelectors) == 0 {
+		return "", nil, defaultPathLimit, errors.NewInternalf(CodeFailBuildJSONPathsQuery, "no field key selectors provided")
+	}
+	from := fmt.Sprintf("%s.%s", DBName, PathTypesTableName)
+
+	// Build a better query using GROUP BY to deduplicate at database level
+	// This aggregates all types per path and gets the max last_seen, then applies LIMIT
+	sb := sqlbuilder.Select(
+		"path",
+		"groupArray(DISTINCT type) AS types",
+		"max(last_seen) AS last_seen",
+	).From(from)
+
+	limit := 0
+	// Add search filter if provided
+	orClauses := []string{}
+	for _, fieldKeySelector := range fieldKeySelectors {
+		// replace [*] with []
+		fieldKeySelector.Name = strings.ReplaceAll(fieldKeySelector.Name, telemetrylogs.ArrayAnyIndex, telemetrylogs.ArraySep)
+		// Extract search text for body JSON keys
+		keyName := CleanPathPrefixes(fieldKeySelector.Name)
+		if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
+			orClauses = append(orClauses, sb.Equal("path", keyName))
+		} else {
+			// Pattern matching for metadata API (defaults to LIKE behavior for other operators)
+			orClauses = append(orClauses, sb.Like("path", querybuilder.FormatValueForContains(keyName)))
+			limit += fieldKeySelector.Limit
+		}
+	}
+	sb.Where(sb.Or(orClauses...))
+
+	// Group by path to get unique paths with aggregated types
+	sb.GroupBy("path")
+
+	// Order by max last_seen to get most recent paths first
+	sb.OrderBy("last_seen DESC")
+	if limit == 0 {
+		limit = defaultPathLimit
+	}
+	sb.Limit(limit)
+
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	return query, args, limit, nil
+}
+
+func getJSONPathIndexes(ctx context.Context, telemetryStore telemetrystore.TelemetryStore, path string) ([]telemetrytypes.JSONDataTypeIndex, error) {
+	// return empty slice if path is an array
+	if strings.Contains(path, telemetrylogs.ArraySep) || strings.Contains(path, telemetrylogs.ArrayAnyIndex) {
+		return nil, nil
+	}
+
+	// list indexes for the path
+	indexes, err := ListLogsJSONIndexes(ctx, telemetryStore, path)
+	if err != nil {
+		return nil, err
+	}
+
+	// build a set of indexes
+	cleanIndexes := []telemetrytypes.JSONDataTypeIndex{}
+	for _, index := range indexes {
+		columnExpr, columnType, err := schemamigrator.UnfoldJSONSubColumnIndexExpr(index.Expression)
+		if err != nil {
+			return nil, errors.WrapInternalf(err, CodeFailLoadLogsJSONIndexes, "failed to unfold JSON sub column index expression: %s", index.Expression)
+		}
+
+		jsonDataType, found := telemetrytypes.MappingStringToJSONDataType[columnType]
+		if !found {
+			return nil, errors.NewInternalf(CodeUnknownJSONDataType, "failed to map column type to JSON data type: %s", columnType)
+		}
+
+		if jsonDataType == telemetrytypes.String {
+			cleanIndexes = append(cleanIndexes, telemetrytypes.JSONDataTypeIndex{
+				Type:             telemetrytypes.String,
+				ColumnExpression: columnExpr,
+				IndexExpression:  index.Expression,
+			})
+		} else if strings.HasPrefix(index.Type, "minmax") {
+			cleanIndexes = append(cleanIndexes, telemetrytypes.JSONDataTypeIndex{
+				Type:             jsonDataType,
+				ColumnExpression: columnExpr,
+				IndexExpression:  index.Expression,
+			})
+		}
+	}
+
+	return cleanIndexes, nil
+}
+
+func buildListLogsJSONIndexesQuery(cluster string, filters ...string) (string, []any) {
+	// Build a better query using GROUP BY to deduplicate at database level
+	// 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))
+
+	sb.Where(sb.Equal("database", telemetrylogs.DBName))
+	sb.Where(sb.Equal("table", telemetrylogs.LogsV2LocalTableName))
+	sb.Where(sb.Or(
+		sb.ILike("expr", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyJSONColumnPrefix))),
+		sb.ILike("expr", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyPromotedColumnPrefix))),
+	))
+
+	filterExprs := []string{}
+	for _, filter := range filters {
+		filterExprs = append(filterExprs, sb.ILike("expr", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(filter))))
+	}
+	sb.Where(sb.Or(filterExprs...))
+
+	return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+}
+
+func ListLogsJSONIndexes(ctx context.Context, telemetryStore telemetrystore.TelemetryStore, filters ...string) ([]schemamigrator.Index, error) {
+	query, args := buildListLogsJSONIndexesQuery(telemetryStore.Cluster(), filters...)
+	rows, err := telemetryStore.ClickhouseDB().Query(ctx, query, args...)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, CodeFailLoadLogsJSONIndexes, "failed to load string indexed columns")
+	}
+	defer rows.Close()
+
+	indexes := []schemamigrator.Index{}
+	for rows.Next() {
+		var name string
+		var typeFull string
+		var expr string
+		var granularity uint64
+		if err := rows.Scan(&name, &typeFull, &expr, &granularity); err != nil {
+			return nil, errors.WrapInternalf(err, CodeFailLoadLogsJSONIndexes, "failed to scan string indexed column")
+		}
+		indexes = append(indexes, schemamigrator.Index{
+			Name:        name,
+			Type:        typeFull,
+			Expression:  expr,
+			Granularity: int(granularity),
+		})
+	}
+
+	return indexes, 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)
+	if err != nil {
+		return nil, errors.WrapInternalf(err, CodeFailLoadPromotedPaths, "failed to load promoted paths")
+	}
+	defer rows.Close()
+
+	next := make(map[string]struct{})
+	for rows.Next() {
+		var path string
+		if err := rows.Scan(&path); err != nil {
+			return nil, errors.WrapInternalf(err, CodeFailLoadPromotedPaths, "failed to scan promoted path")
+		}
+		next[path] = struct{}{}
+	}
+
+	return next, nil
+}
+
+func ListJSONValues(ctx context.Context, conn clickhouse.Conn, 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)
+	if err != nil {
+		return nil, false, err
+	}
+
+	if promoted {
+		path = telemetrylogs.BodyPromotedColumnPrefix + path
+	} else {
+		path = telemetrylogs.BodyJSONColumnPrefix + path
+	}
+
+	from := fmt.Sprintf("%s.%s", telemetrylogs.DBName, telemetrylogs.LogsV2TableName)
+	colExpr := func(typ telemetrytypes.JSONDataType) string {
+		return fmt.Sprintf("dynamicElement(%s, '%s')", path, typ.StringValue())
+	}
+
+	sb := sqlbuilder.Select(
+		colExpr(telemetrytypes.String),
+		colExpr(telemetrytypes.Int64),
+		colExpr(telemetrytypes.Float64),
+		colExpr(telemetrytypes.Bool),
+		colExpr(telemetrytypes.ArrayString),
+		colExpr(telemetrytypes.ArrayInt64),
+		colExpr(telemetrytypes.ArrayFloat64),
+		colExpr(telemetrytypes.ArrayBool),
+		colExpr(telemetrytypes.ArrayDynamic),
+	).From(from)
+	sb.Where(fmt.Sprintf("%s IS NOT NULL", path))
+	sb.Limit(limit)
+
+	contextWithTimeout, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
+	rows, err := conn.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")
+		}
+		return nil, false, errors.WrapInternalf(err, CodeFailListJSONValues, "failed to list JSON values")
+	}
+	defer rows.Close()
+
+	// Get column types to determine proper scan types
+	colTypes := rows.ColumnTypes()
+	scanTargets := make([]any, len(colTypes))
+	for i := range colTypes {
+		scanTargets[i] = reflect.New(colTypes[i].ScanType()).Interface()
+	}
+
+	values := &telemetrytypes.TelemetryFieldValues{}
+	for rows.Next() {
+		// Create fresh scan targets for each row
+		scan := make([]any, len(colTypes))
+		for i := range colTypes {
+			scan[i] = reflect.New(colTypes[i].ScanType()).Interface()
+		}
+
+		if err := rows.Scan(scan...); err != nil {
+			return nil, false, errors.WrapInternalf(err, CodeFailListJSONValues, "failed to scan JSON value row")
+		}
+
+		// Extract values from scan targets and process them
+		// Column order: String, Int64, Float64, Bool, ArrayString, ArrayInt64, ArrayFloat64, ArrayBool, ArrayDynamic
+		var consume func(scan []any) error
+		consume = func(scan []any) error {
+			for _, value := range scan {
+				value := derefValue(value) // dereference the double pointer if it is a pointer
+				switch value := value.(type) {
+				case string:
+					values.StringValues = append(values.StringValues, value)
+				case int64:
+					values.NumberValues = append(values.NumberValues, float64(value))
+				case float64:
+					values.NumberValues = append(values.NumberValues, value)
+				case bool:
+					values.BoolValues = append(values.BoolValues, value)
+				case []*string:
+					for _, str := range value {
+						values.StringValues = append(values.StringValues, *str)
+					}
+				case []*int64:
+					for _, num := range value {
+						values.NumberValues = append(values.NumberValues, float64(*num))
+					}
+				case []*float64:
+					for _, num := range value {
+						values.NumberValues = append(values.NumberValues, float64(*num))
+					}
+				case []*bool:
+					for _, boolVal := range value {
+						values.BoolValues = append(values.BoolValues, *boolVal)
+					}
+				case chcol.Variant:
+					if !value.Nil() {
+						if err := consume([]any{value.Any()}); err != nil {
+							return err
+						}
+					}
+				case []chcol.Variant:
+					extractedValues := make([]any, len(value))
+					for _, variant := range value {
+						if !variant.Nil() && variant.Type() != "JSON" { // skip JSON values cuz they're relevant for nested keys
+							extractedValues = append(extractedValues, variant.Any())
+						}
+					}
+					if err := consume(extractedValues); err != nil {
+						return err
+					}
+				default:
+					if value == nil {
+						continue
+					}
+					return errors.NewInternalf(CodeFailScanJSONValue, "unknown JSON value type: %T", value)
+				}
+			}
+
+			return nil
+		}
+		if err := consume(scan); err != nil {
+			return nil, false, err
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, false, errors.WrapInternalf(err, CodeFailListJSONValues, "error iterating JSON values")
+	}
+
+	return values, true, nil
+}
+
+func derefValue(v any) any {
+	if v == nil {
+		return nil
+	}
+
+	val := reflect.ValueOf(v)
+	for val.Kind() == reflect.Ptr {
+		if val.IsNil() {
+			return nil
+		}
+		val = val.Elem()
+	}
+
+	return val.Interface()
+}
+
+// IsPathPromoted checks if a specific path is promoted
+func IsPathPromoted(ctx context.Context, conn clickhouse.Conn, 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])
+	if err != nil {
+		return false, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to check if path %s is promoted", path)
+	}
+	defer rows.Close()
+
+	return rows.Next(), nil
+}
+
+// TODO(Piyush): Remove this function
+func CleanPathPrefixes(path string) string {
+	path = strings.TrimPrefix(path, telemetrytypes.BodyJSONStringSearchPrefix)
+	path = strings.TrimPrefix(path, telemetrylogs.BodyJSONColumnPrefix)
+	path = strings.TrimPrefix(path, telemetrylogs.BodyPromotedColumnPrefix)
+	return path
+}

pkg/telemetrymetadata/body_json_metadata_test.go

@@ -0,0 +1,99 @@
+package telemetrymetadata
+
+import (
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildGetBodyJSONPathsQuery(t *testing.T) {
+	testCases := []struct {
+		name              string
+		fieldKeySelectors []*telemetrytypes.FieldKeySelector
+		expectedSQL       string
+		expectedArgs      []any
+		expectedLimit     int
+	}{
+
+		{
+			name: "Single search text with EQUAL operator",
+			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
+				{
+					Name:              "user.name",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
+				},
+			},
+			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path = ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
+			expectedArgs:  []any{"user.name", defaultPathLimit},
+			expectedLimit: defaultPathLimit,
+		},
+		{
+			name: "Single search text with LIKE operator",
+			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
+				{
+					Name:              "user",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
+				},
+			},
+			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path LIKE ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
+			expectedArgs:  []any{"user", 100},
+			expectedLimit: 100,
+		},
+		{
+			name: "Multiple search texts with EQUAL operator",
+			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
+				{
+					Name:              "user.name",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
+				},
+				{
+					Name:              "user.age",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
+				},
+			},
+			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path = ? OR path = ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
+			expectedArgs:  []any{"user.name", "user.age", defaultPathLimit},
+			expectedLimit: defaultPathLimit,
+		},
+		{
+			name: "Multiple search texts with LIKE operator",
+			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
+				{
+					Name:              "user",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
+				},
+				{
+					Name:              "admin",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
+				},
+			},
+			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path LIKE ? OR path LIKE ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
+			expectedArgs:  []any{"user", "admin", defaultPathLimit},
+			expectedLimit: defaultPathLimit,
+		},
+		{
+			name: "Search with Contains operator (should default to LIKE)",
+			fieldKeySelectors: []*telemetrytypes.FieldKeySelector{
+				{
+					Name:              "test",
+					SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeFuzzy,
+				},
+			},
+			expectedSQL:   "SELECT path, groupArray(DISTINCT type) AS types, max(last_seen) AS last_seen FROM signoz_metadata.distributed_json_path_types WHERE (path LIKE ?) GROUP BY path ORDER BY last_seen DESC LIMIT ?",
+			expectedArgs:  []any{"test", defaultPathLimit},
+			expectedLimit: defaultPathLimit,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			query, args, limit, err := buildGetBodyJSONPathsQuery(tc.fieldKeySelectors)
+			require.NoError(t, err, "Error building query: %v", err)
+
+			require.Equal(t, tc.expectedSQL, query)
+			require.Equal(t, tc.expectedArgs, args)
+			require.Equal(t, tc.expectedLimit, limit)
+		})
+	}
+}

pkg/telemetrymetadata/tables.go

@@ -1,7 +1,12 @@
 package telemetrymetadata
 
+import otelcollectorconst "github.com/SigNoz/signoz-otel-collector/constants"
+
 const (
 	DBName                           = "signoz_metadata"
 	AttributesMetadataTableName      = "distributed_attributes_metadata"
 	AttributesMetadataLocalTableName = "attributes_metadata"
+	PathTypesTableName               = otelcollectorconst.DistributedPathTypesTable
+	PromotedPathsTableName           = otelcollectorconst.DistributedPromotedPathsTable
+	SkipIndexTableName               = "system.data_skipping_indices"
 )

pkg/telemetrytraces/condition_builder.go

@@ -162,13 +162,15 @@ func (c *conditionBuilder) conditionFor(
 	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
 
 		var value any
-		switch column.Type {
-		case schema.JSONColumnType{}:
+		// schema.JSONColumnType{} now can not be used in switch cases, so we need to check if the column is a JSON column
+		if column.IsJSONColumn() {
 			if operator == qbtypes.FilterOperatorExists {
 				return sb.IsNotNull(tblFieldName), nil
 			} else {
 				return sb.IsNull(tblFieldName), nil
 			}
+		}
+		switch column.Type {
 		case schema.ColumnTypeString,
 			schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 			schema.FixedStringColumnType{Length: 32},
@@ -223,8 +225,8 @@ func (c *conditionBuilder) ConditionFor(
 	operator qbtypes.FilterOperator,
 	value any,
 	sb *sqlbuilder.SelectBuilder,
-    startNs uint64,
-    _ uint64,
+	startNs uint64,
+	_ uint64,
 ) (string, error) {
 	if c.isSpanScopeField(key.Name) {
 		return c.buildSpanScopeCondition(key, operator, value, startNs)

pkg/telemetrytraces/field_mapper.go

@@ -236,8 +236,8 @@ func (m *defaultFieldMapper) FieldFor(
 		return "", err
 	}
 
-	switch column.Type {
-	case schema.JSONColumnType{}:
+	// schema.JSONColumnType{} now can not be used in switch cases, so we need to check if the column is a JSON column
+	if column.IsJSONColumn() {
 		// json is only supported for resource context as of now
 		if key.FieldContext != telemetrytypes.FieldContextResource {
 			return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns, got %s", key.FieldContext.String)
@@ -253,7 +253,9 @@ func (m *defaultFieldMapper) FieldFor(
 		} else {
 			return fmt.Sprintf("multiIf(%s.`%s` IS NOT NULL, %s.`%s`::String, mapContains(%s, '%s'), %s, NULL)", column.Name, key.Name, column.Name, key.Name, oldColumn.Name, key.Name, oldKeyName), nil
 		}
+	}
 
+	switch column.Type {
 	case schema.ColumnTypeString,
 		schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 		schema.ColumnTypeUInt64,

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/field.go

@@ -17,6 +17,10 @@ var (
 	FieldSelectorMatchTypeFuzzy = FieldSelectorMatchType{valuer.NewString("fuzzy")}
 )
 
+// BodyJSONStringSearchPrefix is the prefix used for body JSON search queries
+// e.g., "body.status" where "body." is the prefix
+const BodyJSONStringSearchPrefix = `body.`
+
 type TelemetryFieldKey struct {
 	Name          string        `json:"name"`
 	Description   string        `json:"description,omitempty"`
@@ -24,7 +28,10 @@ type TelemetryFieldKey struct {
 	Signal        Signal        `json:"signal,omitempty"`
 	FieldContext  FieldContext  `json:"fieldContext,omitempty"`
 	FieldDataType FieldDataType `json:"fieldDataType,omitempty"`
-	Materialized  bool          `json:"-"`
+
+	JSONDataType *JSONDataType       `json:"-,omitempty"`
+	Indexes      []JSONDataTypeIndex `json:"-"`
+	Materialized bool                `json:"-"` // refers to promoted in case of body.... fields
 }
 
 func (f TelemetryFieldKey) String() string {

pkg/types/telemetrytypes/field_context.go

@@ -36,6 +36,9 @@ import (
 //
 // - Use `log.` for explicit log context
 //   - `log.severity_text` will always resolve to `severity_text` of log record
+//
+// - Use `body.` to indicate and enforce body context
+//   - `body.key` will look for `key` in the body field
 type FieldContext struct {
 	valuer.String
 }
@@ -49,6 +52,7 @@ var (
 	FieldContextScope       = FieldContext{valuer.NewString("scope")}
 	FieldContextAttribute   = FieldContext{valuer.NewString("attribute")}
 	FieldContextEvent       = FieldContext{valuer.NewString("event")}
+	FieldContextBody        = FieldContext{valuer.NewString("body")}
 	FieldContextUnspecified = FieldContext{valuer.NewString("")}
 
 	// Map string representations to FieldContext values
@@ -65,6 +69,7 @@ var (
 		"point":      FieldContextAttribute,
 		"attribute":  FieldContextAttribute,
 		"event":      FieldContextEvent,
+		"body":       FieldContextBody,
 		"spanfield":  FieldContextSpan,
 		"span":       FieldContextSpan,
 		"logfield":   FieldContextLog,
@@ -144,6 +149,8 @@ func (f FieldContext) TagType() string {
 		return "metricfield"
 	case FieldContextEvent:
 		return "eventfield"
+	case FieldContextBody:
+		return "body"
 	}
 	return ""
 }

pkg/types/telemetrytypes/field_datatype.go

@@ -31,6 +31,9 @@ var (
 	FieldDataTypeArrayInt64  = FieldDataType{valuer.NewString("[]int64")}
 	FieldDataTypeArrayNumber = FieldDataType{valuer.NewString("[]number")}
 
+	FieldDataTypeArrayObject  = FieldDataType{valuer.NewString("[]object")}
+	FieldDataTypeArrayDynamic = FieldDataType{valuer.NewString("[]dynamic")}
+
 	// Map string representations to FieldDataType values
 	// We want to handle all the possible string representations of the data types.
 	// Even if the user uses some non-standard representation, we want to be able to

pkg/types/telemetrytypes/field_test.go

@@ -1,6 +1,7 @@
 package telemetrytypes
 
 import (
+	"reflect"
 	"testing"
 )
 
@@ -86,7 +87,7 @@ func TestGetFieldKeyFromKeyText(t *testing.T) {
 
 	for _, testCase := range testCases {
 		result := GetFieldKeyFromKeyText(testCase.keyText)
-		if result != testCase.expected {
+		if !reflect.DeepEqual(result, testCase.expected) {
 			t.Errorf("expected %v, got %v", testCase.expected, result)
 		}
 	}

pkg/types/telemetrytypes/json_access_plan.go

@@ -0,0 +1,78 @@
+package telemetrytypes
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/SigNoz/signoz-otel-collector/exporter/jsontypeexporter"
+)
+
+type JSONAccessBranchType string
+type JSONAccessPlan = []*JSONAccessNode
+
+const (
+	BranchJSON    JSONAccessBranchType = "json"
+	BranchDynamic JSONAccessBranchType = "dynamic"
+)
+
+type TerminalConfig struct {
+	Key       *TelemetryFieldKey
+	ElemType  JSONDataType
+	ValueType JSONDataType
+}
+
+// Node is now a tree structure representing the complete JSON path traversal
+// that precomputes all possible branches and their types
+type JSONAccessNode struct {
+	// Node information
+	Name       string
+	IsTerminal bool
+	isRoot     bool // marked true for only body_json and body_json_promoted
+
+	// Precomputed type information (single source of truth)
+	AvailableTypes []JSONDataType
+
+	// Array type branches (Array(JSON) vs Array(Dynamic))
+	Branches map[JSONAccessBranchType]*JSONAccessNode
+
+	// Terminal configuration
+	TerminalConfig *TerminalConfig
+
+	// Parent reference for traversal
+	Parent *JSONAccessNode
+
+	// JSON progression parameters (precomputed during planning)
+	MaxDynamicTypes int
+	MaxDynamicPaths int
+}
+
+func NewRootJSONAccessNode(name string, maxDynamicTypes, maxDynamicPaths int) *JSONAccessNode {
+	return &JSONAccessNode{
+		Name:            name,
+		isRoot:          true,
+		MaxDynamicTypes: maxDynamicTypes,
+		MaxDynamicPaths: maxDynamicPaths,
+	}
+}
+
+func (n *JSONAccessNode) Alias() string {
+	if n.isRoot {
+		return n.Name
+	} else if n.Parent == nil {
+		return fmt.Sprintf("`%s`", n.Name)
+	}
+
+	parentAlias := strings.TrimLeft(n.Parent.Alias(), "`")
+	parentAlias = strings.TrimRight(parentAlias, "`")
+
+	sep := jsontypeexporter.ArraySeparator
+	if n.Parent.isRoot {
+		sep = "."
+	}
+	return fmt.Sprintf("`%s%s%s`", parentAlias, sep, n.Name)
+}
+
+func (n *JSONAccessNode) FieldPath() string {
+	key := "`" + n.Name + "`"
+	return n.Parent.Alias() + "." + key
+}

pkg/types/telemetrytypes/json_datatype.go

@@ -0,0 +1,80 @@
+package telemetrytypes
+
+type JSONDataTypeIndex struct {
+	Type             JSONDataType
+	ColumnExpression string
+	IndexExpression  string
+}
+
+type JSONDataType struct {
+	str            string // Store the correct case for ClickHouse
+	IsArray        bool
+	ScalerType     string
+	IndexSupported bool
+}
+
+// Override StringValue to return the correct case
+func (jdt JSONDataType) StringValue() string {
+	return jdt.str
+}
+
+var (
+	String       = JSONDataType{"String", false, "", true}
+	Int64        = JSONDataType{"Int64", false, "", true}
+	Float64      = JSONDataType{"Float64", false, "", true}
+	Bool         = JSONDataType{"Bool", false, "", false}
+	Dynamic      = JSONDataType{"Dynamic", false, "", false}
+	ArrayString  = JSONDataType{"Array(Nullable(String))", true, "String", false}
+	ArrayInt64   = JSONDataType{"Array(Nullable(Int64))", true, "Int64", false}
+	ArrayFloat64 = JSONDataType{"Array(Nullable(Float64))", true, "Float64", false}
+	ArrayBool    = JSONDataType{"Array(Nullable(Bool))", true, "Bool", false}
+	ArrayDynamic = JSONDataType{"Array(Dynamic)", true, "Dynamic", false}
+	ArrayJSON    = JSONDataType{"Array(JSON)", true, "JSON", false}
+)
+
+var MappingStringToJSONDataType = map[string]JSONDataType{
+	"String":                   String,
+	"Int64":                    Int64,
+	"Float64":                  Float64,
+	"Bool":                     Bool,
+	"Dynamic":                  Dynamic,
+	"Array(Nullable(String))":  ArrayString,
+	"Array(Nullable(Int64))":   ArrayInt64,
+	"Array(Nullable(Float64))": ArrayFloat64,
+	"Array(Nullable(Bool))":    ArrayBool,
+	"Array(Dynamic)":           ArrayDynamic,
+	"Array(JSON)":              ArrayJSON,
+}
+
+var ScalerTypeToArrayType = map[JSONDataType]JSONDataType{
+	String:  ArrayString,
+	Int64:   ArrayInt64,
+	Float64: ArrayFloat64,
+	Bool:    ArrayBool,
+	Dynamic: ArrayDynamic,
+}
+
+var MappingFieldDataTypeToJSONDataType = map[FieldDataType]JSONDataType{
+	FieldDataTypeString:       String,
+	FieldDataTypeInt64:        Int64,
+	FieldDataTypeFloat64:      Float64,
+	FieldDataTypeNumber:       Float64,
+	FieldDataTypeBool:         Bool,
+	FieldDataTypeArrayString:  ArrayString,
+	FieldDataTypeArrayInt64:   ArrayInt64,
+	FieldDataTypeArrayFloat64: ArrayFloat64,
+	FieldDataTypeArrayBool:    ArrayBool,
+}
+
+var MappingJSONDataTypeToFieldDataType = map[JSONDataType]FieldDataType{
+	String:       FieldDataTypeString,
+	Int64:        FieldDataTypeInt64,
+	Float64:      FieldDataTypeFloat64,
+	Bool:         FieldDataTypeBool,
+	ArrayString:  FieldDataTypeArrayString,
+	ArrayInt64:   FieldDataTypeArrayInt64,
+	ArrayFloat64: FieldDataTypeArrayFloat64,
+	ArrayBool:    FieldDataTypeArrayBool,
+	ArrayDynamic: FieldDataTypeArrayDynamic,
+	ArrayJSON:    FieldDataTypeArrayObject,
+}