diff --git a/api/seqproxyapi/v1/seq_proxy_api.proto b/api/seqproxyapi/v1/seq_proxy_api.proto index 143d0fec..0c23f945 100644 --- a/api/seqproxyapi/v1/seq_proxy_api.proto +++ b/api/seqproxyapi/v1/seq_proxy_api.proto @@ -491,7 +491,7 @@ message Typing { enum DataType { DATA_TYPE_BYTES = 0; DATA_TYPE_SEQ_ID = 1; - DATA_TYPE_RAW_DOCUMENT = 2; + DATA_TYPE_JSON = 2; DATA_TYPE_STRING = 3; DATA_TYPE_UINT32 = 4; DATA_TYPE_UINT64 = 5; diff --git a/api/storeapi/store_api.proto b/api/storeapi/store_api.proto index f136da7c..bb4d7ddd 100644 --- a/api/storeapi/store_api.proto +++ b/api/storeapi/store_api.proto @@ -321,7 +321,7 @@ message Typing { enum DataType { DATA_TYPE_BYTES = 0; DATA_TYPE_SEQ_ID = 1; - DATA_TYPE_RAW_DOCUMENT = 2; + DATA_TYPE_JSON = 2; DATA_TYPE_STRING = 3; DATA_TYPE_UINT32 = 4; DATA_TYPE_UINT64 = 5; diff --git a/go.mod b/go.mod index 47621de7..ddb5f118 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/aws/aws-sdk-go-v2/credentials v1.20.4 github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.19.13 github.com/aws/aws-sdk-go-v2/service/s3 v1.113.1 + github.com/buger/jsonparser v1.6.1 github.com/c2h5oh/datasize v0.0.0-20200112174442-28bbd4740fee github.com/cep21/circuit/v3 v3.2.2 github.com/felixge/fgprof v0.9.5 diff --git a/go.sum b/go.sum index d03eae42..8f1f3ba9 100644 --- a/go.sum +++ b/go.sum @@ -75,6 +75,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bits-and-blooms/bitset v1.24.2 h1:M7/NzVbsytmtfHbumG+K2bremQPMJuqv1JD3vOaFxp0= github.com/bits-and-blooms/bitset v1.24.2/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= +github.com/buger/jsonparser v1.6.1 h1:I0phFv0PlbLHnM7TZAVjZ2MJ2/eWRTDyuO7GLR98IEs= +github.com/buger/jsonparser v1.6.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= github.com/c2h5oh/datasize v0.0.0-20200112174442-28bbd4740fee h1:BnPxIde0gjtTnc9Er7cxvBk8DHLWhEux0SxayC8dP6I= github.com/c2h5oh/datasize v0.0.0-20200112174442-28bbd4740fee/go.mod h1:S/7n9copUssQ56c7aAgHqftWO4LTf4xY6CGWt8Bc+3M= github.com/cactus/go-statsd-client v3.1.1+incompatible/go.mod h1:cMRcwZDklk7hXp+Law83urTHUiHMzCev/r4JMYr/zU0= diff --git a/pkg/seqproxyapi/v1/mappings.go b/pkg/seqproxyapi/v1/mappings.go index a907764f..23ddb94b 100644 --- a/pkg/seqproxyapi/v1/mappings.go +++ b/pkg/seqproxyapi/v1/mappings.go @@ -133,7 +133,7 @@ func AsyncSearchStatusFromString(s string) (AsyncSearchStatus, error) { var typeMappings = []DataType{ query.DataTypeBytes: DataType_DATA_TYPE_BYTES, query.DataTypeSeqID: DataType_DATA_TYPE_SEQ_ID, - query.DataTypeDocument: DataType_DATA_TYPE_RAW_DOCUMENT, + query.DataTypeJSON: DataType_DATA_TYPE_JSON, query.DataTypeString: DataType_DATA_TYPE_STRING, query.DataTypeUint32: DataType_DATA_TYPE_UINT32, query.DataTypeUint64: DataType_DATA_TYPE_UINT64, diff --git a/pkg/seqproxyapi/v1/seq_proxy_api.pb.go b/pkg/seqproxyapi/v1/seq_proxy_api.pb.go index 68464be6..dc9d7ef4 100644 --- a/pkg/seqproxyapi/v1/seq_proxy_api.pb.go +++ b/pkg/seqproxyapi/v1/seq_proxy_api.pb.go @@ -298,7 +298,7 @@ type DataType int32 const ( DataType_DATA_TYPE_BYTES DataType = 0 DataType_DATA_TYPE_SEQ_ID DataType = 1 - DataType_DATA_TYPE_RAW_DOCUMENT DataType = 2 + DataType_DATA_TYPE_JSON DataType = 2 DataType_DATA_TYPE_STRING DataType = 3 DataType_DATA_TYPE_UINT32 DataType = 4 DataType_DATA_TYPE_UINT64 DataType = 5 @@ -314,7 +314,7 @@ var ( DataType_name = map[int32]string{ 0: "DATA_TYPE_BYTES", 1: "DATA_TYPE_SEQ_ID", - 2: "DATA_TYPE_RAW_DOCUMENT", + 2: "DATA_TYPE_JSON", 3: "DATA_TYPE_STRING", 4: "DATA_TYPE_UINT32", 5: "DATA_TYPE_UINT64", @@ -327,7 +327,7 @@ var ( DataType_value = map[string]int32{ "DATA_TYPE_BYTES": 0, "DATA_TYPE_SEQ_ID": 1, - "DATA_TYPE_RAW_DOCUMENT": 2, + "DATA_TYPE_JSON": 2, "DATA_TYPE_STRING": 3, "DATA_TYPE_UINT32": 4, "DATA_TYPE_UINT64": 5, @@ -3705,11 +3705,11 @@ const file_seqproxyapi_v1_seq_proxy_api_proto_rawDesc = "" + "\x16AsyncSearchStatusError\x10\x03*L\n" + "\rControlAction\x12\x1e\n" + "\x1aCONTROL_ACTION_UNSPECIFIED\x10\x00\x12\x1b\n" + - "\x17CONTROL_ACTION_FINALIZE\x10\x01*\x8d\x02\n" + + "\x17CONTROL_ACTION_FINALIZE\x10\x01*\x85\x02\n" + "\bDataType\x12\x13\n" + "\x0fDATA_TYPE_BYTES\x10\x00\x12\x14\n" + - "\x10DATA_TYPE_SEQ_ID\x10\x01\x12\x1a\n" + - "\x16DATA_TYPE_RAW_DOCUMENT\x10\x02\x12\x14\n" + + "\x10DATA_TYPE_SEQ_ID\x10\x01\x12\x12\n" + + "\x0eDATA_TYPE_JSON\x10\x02\x12\x14\n" + "\x10DATA_TYPE_STRING\x10\x03\x12\x14\n" + "\x10DATA_TYPE_UINT32\x10\x04\x12\x14\n" + "\x10DATA_TYPE_UINT64\x10\x05\x12\x13\n" + diff --git a/pkg/seqproxyapi/v1/stream_render.go b/pkg/seqproxyapi/v1/stream_render.go index 447dce79..ab4be070 100644 --- a/pkg/seqproxyapi/v1/stream_render.go +++ b/pkg/seqproxyapi/v1/stream_render.go @@ -135,7 +135,7 @@ func (t *Typing) renderCell(cell []byte) any { return string(cell) case DataType_DATA_TYPE_BYTES: return cell - case DataType_DATA_TYPE_RAW_DOCUMENT: + case DataType_DATA_TYPE_JSON: return json.RawMessage(cell) case DataType_DATA_TYPE_UINT32: return encoding.Uint32FromBytes(cell) diff --git a/pkg/seqproxyapi/v1/stream_render_test.go b/pkg/seqproxyapi/v1/stream_render_test.go index 0741b4f7..3eeb5ca7 100644 --- a/pkg/seqproxyapi/v1/stream_render_test.go +++ b/pkg/seqproxyapi/v1/stream_render_test.go @@ -31,7 +31,7 @@ func TestRowSchemaRender(t *testing.T) { []*Typing{ {Title: "id", Type: DataType_DATA_TYPE_SEQ_ID}, {Title: "time", Type: DataType_DATA_TYPE_UINT64}, - {Title: "data", Type: DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "data", Type: DataType_DATA_TYPE_JSON}, }, [][]byte{ []byte("46e48be997010000-e70163d0fa7582e4"), diff --git a/pkg/storeapi/mappings.go b/pkg/storeapi/mappings.go index 29236007..d981cd85 100644 --- a/pkg/storeapi/mappings.go +++ b/pkg/storeapi/mappings.go @@ -147,7 +147,7 @@ func MustProtoAsyncSearchStatus(s asyncsearcher.AsyncSearchStatus) AsyncSearchSt var typeMappings = []DataType{ query.DataTypeBytes: DataType_DATA_TYPE_BYTES, query.DataTypeSeqID: DataType_DATA_TYPE_SEQ_ID, - query.DataTypeDocument: DataType_DATA_TYPE_RAW_DOCUMENT, + query.DataTypeJSON: DataType_DATA_TYPE_JSON, query.DataTypeString: DataType_DATA_TYPE_STRING, query.DataTypeUint32: DataType_DATA_TYPE_UINT32, query.DataTypeUint64: DataType_DATA_TYPE_UINT64, diff --git a/pkg/storeapi/store_api.pb.go b/pkg/storeapi/store_api.pb.go index ab21f03d..b076cee4 100644 --- a/pkg/storeapi/store_api.pb.go +++ b/pkg/storeapi/store_api.pb.go @@ -301,7 +301,7 @@ type DataType int32 const ( DataType_DATA_TYPE_BYTES DataType = 0 DataType_DATA_TYPE_SEQ_ID DataType = 1 - DataType_DATA_TYPE_RAW_DOCUMENT DataType = 2 + DataType_DATA_TYPE_JSON DataType = 2 DataType_DATA_TYPE_STRING DataType = 3 DataType_DATA_TYPE_UINT32 DataType = 4 DataType_DATA_TYPE_UINT64 DataType = 5 @@ -317,7 +317,7 @@ var ( DataType_name = map[int32]string{ 0: "DATA_TYPE_BYTES", 1: "DATA_TYPE_SEQ_ID", - 2: "DATA_TYPE_RAW_DOCUMENT", + 2: "DATA_TYPE_JSON", 3: "DATA_TYPE_STRING", 4: "DATA_TYPE_UINT32", 5: "DATA_TYPE_UINT64", @@ -330,7 +330,7 @@ var ( DataType_value = map[string]int32{ "DATA_TYPE_BYTES": 0, "DATA_TYPE_SEQ_ID": 1, - "DATA_TYPE_RAW_DOCUMENT": 2, + "DATA_TYPE_JSON": 2, "DATA_TYPE_STRING": 3, "DATA_TYPE_UINT32": 4, "DATA_TYPE_UINT64": 5, @@ -3255,11 +3255,11 @@ const file_storeapi_store_api_proto_rawDesc = "" + "\x16AsyncSearchStatusError\x10\x03*L\n" + "\rControlAction\x12\x1e\n" + "\x1aCONTROL_ACTION_UNSPECIFIED\x10\x00\x12\x1b\n" + - "\x17CONTROL_ACTION_FINALIZE\x10\x01*\x8d\x02\n" + + "\x17CONTROL_ACTION_FINALIZE\x10\x01*\x85\x02\n" + "\bDataType\x12\x13\n" + "\x0fDATA_TYPE_BYTES\x10\x00\x12\x14\n" + - "\x10DATA_TYPE_SEQ_ID\x10\x01\x12\x1a\n" + - "\x16DATA_TYPE_RAW_DOCUMENT\x10\x02\x12\x14\n" + + "\x10DATA_TYPE_SEQ_ID\x10\x01\x12\x12\n" + + "\x0eDATA_TYPE_JSON\x10\x02\x12\x14\n" + "\x10DATA_TYPE_STRING\x10\x03\x12\x14\n" + "\x10DATA_TYPE_UINT32\x10\x04\x12\x14\n" + "\x10DATA_TYPE_UINT64\x10\x05\x12\x13\n" + diff --git a/proxy/search/stream_search_test.go b/proxy/search/stream_search_test.go index 846472f4..71d3069d 100644 --- a/proxy/search/stream_search_test.go +++ b/proxy/search/stream_search_test.go @@ -7,7 +7,6 @@ import ( "sync" "testing" - insaneJSON "github.com/ozontech/insane-json" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc/metadata" @@ -189,7 +188,7 @@ func TestNewStreamSearchIteratorEOFOnPrefetch(t *testing.T) { func testTyping() []*storeapi.Typing { return []*storeapi.Typing{ {Title: "id", Type: storeapi.DataType_DATA_TYPE_SEQ_ID}, - {Title: "data", Type: storeapi.DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "data", Type: storeapi.DataType_DATA_TYPE_JSON}, } } @@ -314,15 +313,15 @@ func (f *fakeStream) RecvMsg(any) error { return nil } func TestSchemaFromTyping(t *testing.T) { typing := []*storeapi.Typing{ {Title: "id", Type: storeapi.DataType_DATA_TYPE_SEQ_ID}, - {Title: "data", Type: storeapi.DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "data", Type: storeapi.DataType_DATA_TYPE_JSON}, } schema, types, err := schemaFromTyping(typing) require.NoError(t, err) assert.Equal(t, 2, schema.Len()) - assert.Equal(t, []query.DataType{query.DataTypeSeqID, query.DataTypeDocument}, types) + assert.Equal(t, []query.DataType{query.DataTypeSeqID, query.DataTypeJSON}, types) assert.Equal(t, 0, schema.MustColumn[seq.ID]("id").Idx()) - assert.Equal(t, query.DataTypeDocument, schema.MustColumn[*insaneJSON.Root]("data").DataType()) + assert.Equal(t, query.DataTypeJSON, schema.MustColumn[[]byte]("data").DataType()) } func TestSchemaFromTypingDuplicateName(t *testing.T) { @@ -343,11 +342,11 @@ func TestValidateShardSchemas(t *testing.T) { docsTyping := []*storeapi.Typing{ {Title: "id", Type: storeapi.DataType_DATA_TYPE_SEQ_ID}, - {Title: "data", Type: storeapi.DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "data", Type: storeapi.DataType_DATA_TYPE_JSON}, } otherTyping := []*storeapi.Typing{ {Title: "id", Type: storeapi.DataType_DATA_TYPE_SEQ_ID}, - {Title: "payload", Type: storeapi.DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "payload", Type: storeapi.DataType_DATA_TYPE_JSON}, } aggsTyping := []*storeapi.Typing{ {Title: "token", Type: storeapi.DataType_DATA_TYPE_STRING}, diff --git a/proxyapi/grpc_stream_search.go b/proxyapi/grpc_stream_search.go index eeb8c724..10476756 100644 --- a/proxyapi/grpc_stream_search.go +++ b/proxyapi/grpc_stream_search.go @@ -6,7 +6,6 @@ import ( "fmt" "io" - insaneJSON "github.com/ozontech/insane-json" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -304,7 +303,7 @@ func docsTyping() []*seqproxyapi.Typing { return []*seqproxyapi.Typing{ {Title: "id", Type: seqproxyapi.DataType_DATA_TYPE_SEQ_ID}, {Title: "time", Type: seqproxyapi.DataType_DATA_TYPE_UINT64}, // extract time from id - {Title: "data", Type: seqproxyapi.DataType_DATA_TYPE_RAW_DOCUMENT}, + {Title: "data", Type: seqproxyapi.DataType_DATA_TYPE_JSON}, } } @@ -321,7 +320,7 @@ func schemaToTyping(out *query.Schema) []*seqproxyapi.Typing { var ( docIDCol = query.DocsSchema.MustColumn[seq.ID](query.DocsIDCol) - docDataCol = query.DocsSchema.MustColumn[*insaneJSON.Root](query.DocsDataCol) + docDataCol = query.DocsSchema.MustColumn[[]byte](query.DocsDataCol) ) // converts *query.Record to *seqproxyapi.Record according to hardcoded schemas from both store and proxy diff --git a/query/column.go b/query/column.go index b49be7d2..df60b5e1 100644 --- a/query/column.go +++ b/query/column.go @@ -3,8 +3,6 @@ package query import ( "cmp" - insaneJSON "github.com/ozontech/insane-json" - "github.com/ozontech/seq-db/seq" ) @@ -74,11 +72,11 @@ func StringColumn(idx int) Column[string] { } } -func DocColumn(idx int) Column[*insaneJSON.Root] { - return Column[*insaneJSON.Root]{ +func JSONColumn(idx int) Column[[]byte] { + return Column[[]byte]{ idx: idx, - dataType: DataTypeDocument, - get: (*RecordVals).AsDoc, + dataType: DataTypeJSON, + get: (*RecordVals).AsJSON, } } diff --git a/query/exec/docs_datasource.go b/query/exec/docs_datasource.go index 46db6efe..a50343eb 100644 --- a/query/exec/docs_datasource.go +++ b/query/exec/docs_datasource.go @@ -165,7 +165,7 @@ func makeDocumentRecord(id seq.ID, payload []byte) *query.Record { return &query.Record{ Vals: []*query.RecordVals{ query.NewSeqIDVals(id), - query.NewDocVals(payload), + query.NewJSONVals(payload), }, } } diff --git a/query/exec/extractor.go b/query/exec/extractor.go index 23f0c3bc..b888a31d 100644 --- a/query/exec/extractor.go +++ b/query/exec/extractor.go @@ -1,56 +1,80 @@ package exec import ( - insaneJSON "github.com/ozontech/insane-json" + "slices" + + "github.com/buger/jsonparser" "github.com/ozontech/seq-db/query" ) -type DocFieldsExtractor struct { - input query.RecordProducer - docCol query.Column[*insaneJSON.Root] +var nullFieldValue = []byte("null") - // extractFields lists scalar JSON fields to extract out of the document column as string values. - extractFields []string +type JSONFieldsExtractor struct { + input query.RecordProducer + col query.Column[[]byte] - // roots holds every record whose val has been decoded (spawned an insaneJSON root). - // They are released back to the library pool in Finalize. - roots []*query.Record + extractFields []string + paths [][]string } -func NewDocFieldsExtractor( +func NewJSONFieldsExtractor( input query.RecordProducer, - docCol query.Column[*insaneJSON.Root], + col query.Column[[]byte], extractFields ...string, -) *DocFieldsExtractor { - return &DocFieldsExtractor{ +) *JSONFieldsExtractor { + paths := make([][]string, len(extractFields)) + for i, f := range extractFields { + paths[i] = []string{f} + } + return &JSONFieldsExtractor{ input: input, - docCol: docCol, + col: col, extractFields: extractFields, + paths: paths, } } -func (e *DocFieldsExtractor) Next() *query.Record { +func (e *JSONFieldsExtractor) Next() *query.Record { r := e.input.Next() if r == nil { return nil } - root := e.docCol.Val(r) - for _, field := range e.extractFields { - // now we treat everything as strings, but later will support other types as well - fieldVal := root.Dig(field).AsBytes() - r.Vals = append(r.Vals, query.NewRecordVals(query.DataTypeString, fieldVal)) + doc := e.col.Val(r) + base := len(r.Vals) + r.Vals = slices.Grow(r.Vals, len(e.extractFields)) + + // pre fill vals in case of not found paths + for range e.extractFields { + // now treat everything as strings, but later will support other types as well + r.Vals = append(r.Vals, query.NewRecordVals(query.DataTypeString, nil)) } - e.roots = append(e.roots, r) + + jsonparser.EachKey(doc, func(i int, value []byte, dataType jsonparser.ValueType, _ error) { + if i >= 0 { + r.Vals[base+i] = query.NewRecordVals(query.DataTypeString, jsonFieldValue(value, dataType)) + } + }, e.paths...) return r } -func (e *DocFieldsExtractor) Finalize() *query.Summary { - for _, r := range e.roots { - r.Release() - } - e.roots = nil +func (e *JSONFieldsExtractor) Finalize() *query.Summary { return e.input.Finalize() } + +func jsonFieldValue(value []byte, dataType jsonparser.ValueType) []byte { + switch dataType { + case jsonparser.String: + unescaped, err := jsonparser.Unescape(value, nil) + if err != nil { + return nil + } + return unescaped + case jsonparser.Null: + return nullFieldValue + default: + return value + } +} diff --git a/query/exec/extractor_test.go b/query/exec/extractor_test.go index 501d73b6..f3695c89 100644 --- a/query/exec/extractor_test.go +++ b/query/exec/extractor_test.go @@ -10,15 +10,15 @@ import ( "github.com/ozontech/seq-db/query/encoding" ) -func TestDocFieldsExtractorExtractsScalars(t *testing.T) { +func TestJSONFieldsExtractorExtractsScalars(t *testing.T) { input := &testProducer{data: makeExtractorInputRecords([]string{ `{"service":"svc-1","level":3,"active":true,"absent":null}`, `{"service":"svc-2","level":7,"active":false}`, })} - extractor := NewDocFieldsExtractor( + extractor := NewJSONFieldsExtractor( input, - query.DocColumn(1), + query.JSONColumn(1), "service", "level", "active", "absent", ) @@ -42,12 +42,12 @@ func TestDocFieldsExtractorExtractsScalars(t *testing.T) { assert.Nil(t, extractor.Next()) } -func TestDocFieldsExtractorMissingFieldYieldsEmpty(t *testing.T) { +func TestJSONFieldsExtractorMissingFieldYieldsEmpty(t *testing.T) { input := &testProducer{data: makeExtractorInputRecords([]string{ `{"service":"svc-1"}`, })} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1), "nope") + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), "nope") r := extractor.Next() assert.NotNil(t, r) @@ -55,26 +55,51 @@ func TestDocFieldsExtractorMissingFieldYieldsEmpty(t *testing.T) { assert.Equal(t, "", r.Vals[2].AsString()) } -func TestDocFieldsExtractorObjectAndArrayFields(t *testing.T) { - // This case documents the scalar-only contract: object and array fields yield an - // empty string, not their JSON representation. +func TestJSONFieldsExtractorObjectAndArrayFields(t *testing.T) { input := &testProducer{data: makeExtractorInputRecords([]string{ - `{"service":"svc-1","obj":{"a":1},"arr":[1,2]}`, + `{"service":"svc-1","obj":{"a":1,"b":"x"},"arr":[1,2]}`, })} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1), "obj", "arr") + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), "obj", "arr") r := extractor.Next() assert.NotNil(t, r) - assert.Equal(t, "", r.Vals[2].AsString()) - assert.Equal(t, "", r.Vals[3].AsString()) + assert.JSONEq(t, `{"a":1,"b":"x"}`, r.Vals[2].AsString()) + assert.JSONEq(t, `[1,2]`, r.Vals[3].AsString()) +} + +func TestJSONFieldsExtractorEscapedStrings(t *testing.T) { + input := &testProducer{data: makeExtractorInputRecords([]string{ + `{"msg":"line1\nline2","quote":"say \"hi\""}`, + })} + + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), "msg", "quote") + + r := extractor.Next() + assert.NotNil(t, r) + assert.Equal(t, "line1\nline2", r.Vals[2].AsString()) + assert.Equal(t, `say "hi"`, r.Vals[3].AsString()) +} + +func TestJSONFieldsExtractorKeysWithSpecialChars(t *testing.T) { + input := &testProducer{data: makeExtractorInputRecords([]string{ + `{"http.method":"GET","{a}":1,"a b":" "}`, + })} + + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), "http.method", "{a}", "a b") + + r := extractor.Next() + assert.NotNil(t, r) + assert.Equal(t, "GET", r.Vals[2].AsString()) + assert.Equal(t, "1", r.Vals[3].AsString()) + assert.Equal(t, " ", r.Vals[4].AsString()) } -func TestDocFieldsExtractorNoFields(t *testing.T) { +func TestJSONFieldsExtractorNoFields(t *testing.T) { docs := []string{`{"service":"svc-1"}`, `{"service":"svc-2"}`} input := &testProducer{data: makeExtractorInputRecords(docs)} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1)) + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1)) count := 0 for r := extractor.Next(); r != nil; r = extractor.Next() { @@ -85,29 +110,28 @@ func TestDocFieldsExtractorNoFields(t *testing.T) { assert.Equal(t, 2, count) } -func TestDocFieldsExtractorFinalize(t *testing.T) { +func TestJSONFieldsExtractorFinalize(t *testing.T) { input := &testProducer{data: makeExtractorInputRecords([]string{ `{"service":"svc-1"}`, `{"service":"svc-2"}`, }), total: 2} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1), "service") + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), "service") for r := extractor.Next(); r != nil; r = extractor.Next() { } summary := extractor.Finalize() assert.Equal(t, uint64(2), summary.Total) - assert.Nil(t, extractor.roots) } -func TestDocFieldsExtractorPipeline(t *testing.T) { +func TestJSONFieldsExtractorPipeline(t *testing.T) { // E2E check: a downstream reader uses the extractor's field // values via a string column appended right after the original vals. const field = "k8s_pod" inputData := makeTestInputRecords(10) input := &testProducer{data: inputData} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1), field) + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), field) podCol := query.StringColumn(2) for r := extractor.Next(); r != nil; r = extractor.Next() { @@ -122,7 +146,7 @@ func makeExtractorInputRecords(docs []string) []*query.Record { out = append(out, &query.Record{ Vals: []*query.RecordVals{ query.NewRecordVals(query.DataTypeUint64, encoding.Uint64ToBytes(0)), - query.NewRecordVals(query.DataTypeDocument, []byte(doc)), + query.NewRecordVals(query.DataTypeJSON, []byte(doc)), }, }) } diff --git a/query/exec/limiter_test.go b/query/exec/limiter_test.go index 6742fe00..87450809 100644 --- a/query/exec/limiter_test.go +++ b/query/exec/limiter_test.go @@ -98,7 +98,7 @@ func makeTestInputRecords(count int) []*query.Record { out = append(out, &query.Record{ Vals: []*query.RecordVals{ query.NewUint32Vals(uint32(i)), - query.NewDocVals([]byte(doc)), + query.NewJSONVals([]byte(doc)), }, }) } diff --git a/query/exec/projector.go b/query/exec/projector.go index f2ca3fff..ea266455 100644 --- a/query/exec/projector.go +++ b/query/exec/projector.go @@ -3,7 +3,7 @@ package exec import ( "slices" - insaneJSON "github.com/ozontech/insane-json" + "github.com/buger/jsonparser" "github.com/ozontech/seq-db/query" ) @@ -13,81 +13,92 @@ type FieldsFilter struct { AllowList bool } -type DocProjector struct { +type JSONProjector struct { input query.RecordProducer - col query.Column[*insaneJSON.Root] + col query.Column[[]byte] filter *FieldsFilter - // roots holds every record whose doc val has been decoded (spawned an insaneJSON root). - // They are released back to the library pool in Finalize. - roots []*query.Record + quoted [][]byte // escaped keys + paths [][]string } -func NewDocProjector( +func NewJSONProjector( input query.RecordProducer, - col query.Column[*insaneJSON.Root], + col query.Column[[]byte], filter *FieldsFilter, -) *DocProjector { - return &DocProjector{ +) *JSONProjector { + p := &JSONProjector{ input: input, col: col, filter: filter, } + + if filter.AllowList { + p.paths = make([][]string, len(filter.Fields)) + p.quoted = make([][]byte, len(filter.Fields)) + for i, f := range filter.Fields { + p.paths[i] = []string{f} + p.quoted[i] = jsonparser.Escape(f) + } + } + + return p } -func (p *DocProjector) Next() *query.Record { +func (p *JSONProjector) Next() *query.Record { r := p.input.Next() if r == nil { return nil } - decoder := p.col.Val(r) + doc := p.col.Val(r) - var newRecord *query.Record - if !p.filter.AllowList { - // It is block list, so remove given fields from document. - for _, field := range p.filter.Fields { - decoder.Dig(field).Suicide() - } - newRecord = p.makeRecordWithNewVals(r, decoder.Encode(nil)) + var newDoc []byte + if p.filter.AllowList { + newDoc = p.projectAllowList(doc) } else { - // Keep only given fields. - // fieldsToRemove contains fields that should be removed. - // It is complex to do it in-place because decoder.Suicide makes decoder.AsFields() invalid. - var fieldsToRemove []*insaneJSON.Node - for _, field := range decoder.AsFields() { - fieldName := field.AsString() - if !slices.Contains(p.filter.Fields, fieldName) { - fieldsToRemove = append(fieldsToRemove, field.AsFieldValue()) - } - } - for _, field := range fieldsToRemove { - field.Suicide() + newDoc = doc + for _, f := range p.filter.Fields { + newDoc = jsonparser.Delete(newDoc, f) } - newRecord = p.makeRecordWithNewVals(r, decoder.Encode(nil)) } - p.roots = append(p.roots, r) + vals := slices.Clone(r.Vals) + vals[p.col.Idx()] = query.NewJSONVals(newDoc) - return newRecord + return query.NewRecord(vals) } -func (p *DocProjector) Finalize() *query.Summary { - for _, r := range p.roots { - r.Release() - } - p.roots = nil +func (p *JSONProjector) Finalize() *query.Summary { return p.input.Finalize() } -func (p *DocProjector) makeRecordWithNewVals(old *query.Record, newRawData []byte) *query.Record { - newRecordVals := make([]*query.RecordVals, len(old.Vals)) - for i := range len(old.Vals) { - rawData := old.Vals[i].RawData() - if i == p.col.Idx() { - rawData = newRawData +func (p *JSONProjector) projectAllowList(doc []byte) []byte { + buf := make([]byte, 0, len(doc)) + buf = append(buf, '{') + first := true + + jsonparser.EachKey(doc, func(i int, value []byte, dataType jsonparser.ValueType, _ error) { + if dataType == jsonparser.Unknown || dataType == jsonparser.NotExist { + return } - newRecordVals[i] = query.NewRecordVals(old.Vals[i].Type, rawData) - } - return query.NewRecord(newRecordVals) + + if !first { + buf = append(buf, ',') + } + first = false + + buf = append(buf, p.quoted[i]...) + buf = append(buf, ':') + + if dataType == jsonparser.String { + buf = append(buf, '"') + buf = append(buf, value...) + buf = append(buf, '"') + } else { + buf = append(buf, value...) + } + }, p.paths...) + + return append(buf, '}') } diff --git a/query/exec/projector_test.go b/query/exec/projector_test.go index a9b29eed..9693c65f 100644 --- a/query/exec/projector_test.go +++ b/query/exec/projector_test.go @@ -9,8 +9,8 @@ import ( "github.com/ozontech/seq-db/query" ) -func TestDocProjectorFields(t *testing.T) { - testDocProjector( +func TestJSONProjectorFields(t *testing.T) { + testJSONProjector( t, &FieldsFilter{Fields: []string{"service", "level"}, AllowList: true}, []string{ @@ -21,7 +21,7 @@ func TestDocProjectorFields(t *testing.T) { } func TestProjectorFieldsExcept(t *testing.T) { - testDocProjector( + testJSONProjector( t, &FieldsFilter{Fields: []string{"level"}, AllowList: false}, []string{ @@ -31,13 +31,91 @@ func TestProjectorFieldsExcept(t *testing.T) { ) } -func testDocProjector(t *testing.T, fieldsFilter *FieldsFilter, wantDocs []string) { +func TestJSONProjectorAllowListEscapedAndMissingFields(t *testing.T) { + input := &testProducer{data: []*query.Record{ + makeDocRecord(`{"msg":"line1\nline2","service":"svc-1"}`), + makeDocRecord(`{"service":"svc-2"}`), + }} + + projector := NewJSONProjector(input, query.JSONColumn(1), &FieldsFilter{ + Fields: []string{"service", "msg"}, + AllowList: true, + }) + + r1 := projector.Next() + assert.NotNil(t, r1) + assert.JSONEq(t, `{"service":"svc-1","msg":"line1\nline2"}`, string(r1.Vals[1].RawData())) + + r2 := projector.Next() + assert.NotNil(t, r2) + assert.JSONEq(t, `{"service":"svc-2"}`, string(r2.Vals[1].RawData())) +} + +func TestJSONProjectorAllowListEmpty(t *testing.T) { + input := &testProducer{data: []*query.Record{makeDocRecord(`{"service":"svc-1"}`)}} + + projector := NewJSONProjector(input, query.JSONColumn(1), &FieldsFilter{AllowList: true}) + + r := projector.Next() + assert.NotNil(t, r) + assert.Equal(t, `{}`, string(r.Vals[1].RawData())) +} + +func TestJSONProjectorObjectField(t *testing.T) { + input := &testProducer{data: []*query.Record{makeDocRecord(`{"service":"svc-1","obj":{"a":1},"level":3}`)}} + + projector := NewJSONProjector(input, query.JSONColumn(1), &FieldsFilter{ + Fields: []string{"obj"}, + AllowList: true, + }) + + r := projector.Next() + assert.NotNil(t, r) + assert.JSONEq(t, `{"obj":{"a":1}}`, string(r.Vals[1].RawData())) +} + +func TestJSONProjectorExceptObjectField(t *testing.T) { + input := &testProducer{data: []*query.Record{makeDocRecord(`{"service":"svc-1","obj":{"a":1},"level":3}`)}} + + projector := NewJSONProjector(input, query.JSONColumn(1), &FieldsFilter{ + Fields: []string{"obj"}, + AllowList: false, + }) + + r := projector.Next() + assert.NotNil(t, r) + assert.JSONEq(t, `{"service":"svc-1","level":3}`, string(r.Vals[1].RawData())) +} + +func TestJSONProjectorExceptAbsentField(t *testing.T) { + input := &testProducer{data: []*query.Record{makeDocRecord(`{"service":"svc-1"}`)}} + + projector := NewJSONProjector(input, query.JSONColumn(1), &FieldsFilter{ + Fields: []string{"nope"}, + AllowList: false, + }) + + r := projector.Next() + assert.NotNil(t, r) + assert.JSONEq(t, `{"service":"svc-1"}`, string(r.Vals[1].RawData())) +} + +func makeDocRecord(doc string) *query.Record { + return &query.Record{ + Vals: []*query.RecordVals{ + query.NewUint32Vals(0), + query.NewJSONVals([]byte(doc)), + }, + } +} + +func testJSONProjector(t *testing.T, fieldsFilter *FieldsFilter, wantDocs []string) { t.Helper() inputData := makeTestInputRecords(2) input := testProducer{data: inputData} - projector := NewDocProjector(&input, query.DocColumn(1), fieldsFilter) + projector := NewJSONProjector(&input, query.JSONColumn(1), fieldsFilter) outputData := make([]*query.Record, 0) for r := projector.Next(); r != nil; r = projector.Next() { @@ -52,12 +130,12 @@ func testDocProjector(t *testing.T, fieldsFilter *FieldsFilter, wantDocs []strin assert.Equal(t, wantDocs, outputDocs) } -func TestDocProjectorKeepsUpstreamValsAlive(t *testing.T) { +func TestJSONProjectorKeepsUpstreamValsAlive(t *testing.T) { const field = "k8s_pod" input := &testProducer{data: makeTestInputRecords(10)} - extractor := NewDocFieldsExtractor(input, query.DocColumn(1), field) - projector := NewDocProjector(extractor, query.DocColumn(1), &FieldsFilter{ + extractor := NewJSONFieldsExtractor(input, query.JSONColumn(1), field) + projector := NewJSONProjector(extractor, query.JSONColumn(1), &FieldsFilter{ Fields: []string{"service", "level"}, AllowList: false, }) diff --git a/query/plan/ops.go b/query/plan/ops.go index 95883150..3207104d 100644 --- a/query/plan/ops.go +++ b/query/plan/ops.go @@ -1,8 +1,6 @@ package plan import ( - insaneJSON "github.com/ozontech/insane-json" - "github.com/ozontech/seq-db/parser" "github.com/ozontech/seq-db/query" "github.com/ozontech/seq-db/query/exec" @@ -30,8 +28,8 @@ func (op *ExtractOp) Apply(input query.RecordProducer, in *query.Schema) (query. } out := in.Extend(descs...) - docCol := in.MustColumn[*insaneJSON.Root](op.OpField) - return exec.NewDocFieldsExtractor(input, docCol, op.Fields...), out + col := in.MustColumn[[]byte](op.OpField) + return exec.NewJSONFieldsExtractor(input, col, op.Fields...), out } type FilterOp struct { @@ -52,8 +50,8 @@ type ProjectOp struct { } func (op *ProjectOp) Apply(input query.RecordProducer, in *query.Schema) (query.RecordProducer, *query.Schema) { - docCol := in.MustColumn[*insaneJSON.Root](op.OpField) - return exec.NewDocProjector(input, docCol, &exec.FieldsFilter{ + col := in.MustColumn[[]byte](op.OpField) + return exec.NewJSONProjector(input, col, &exec.FieldsFilter{ Fields: op.Fields, AllowList: op.AllowList, }), in diff --git a/query/record.go b/query/record.go index 1e087131..3b100a99 100644 --- a/query/record.go +++ b/query/record.go @@ -3,13 +3,10 @@ package query import ( "fmt" - insaneJSON "github.com/ozontech/insane-json" - "github.com/ozontech/seq-db/query/encoding" "github.com/ozontech/seq-db/seq" ) -// executors make use of val's index, executor's parameters has colIdx field type Record struct { Vals []*RecordVals } @@ -20,12 +17,6 @@ func NewRecord(vals []*RecordVals) *Record { } } -func (r *Record) Release() { - for _, v := range r.Vals { - v.Release() - } -} - type DataType byte func (t DataType) String() string { @@ -34,8 +25,8 @@ func (t DataType) String() string { return "bytes" case DataTypeSeqID: return "seq_id" - case DataTypeDocument: - return "document" + case DataTypeJSON: + return "json" case DataTypeString: return "string" case DataTypeUint32: @@ -60,7 +51,7 @@ func (t DataType) String() string { const ( DataTypeBytes DataType = iota DataTypeSeqID - DataTypeDocument + DataTypeJSON DataTypeString DataTypeUint32 DataTypeUint64 @@ -74,7 +65,7 @@ const ( type RecordVals struct { Type DataType - // decoded marks whether the doc/f64s/strs cache has been filled. + // decoded marks whether the f64s/strs cache has been filled. // Exactly one cache field is used per val so a single flag covers all types. decoded bool @@ -84,7 +75,6 @@ type RecordVals struct { // Scalars are decoded from rawData on every access — the decode is a // single load, caching it in a struct field costs more (bigger struct, // extra branch) than it saves. - doc *insaneJSON.Root f64s []float64 strs []string } @@ -140,11 +130,10 @@ func NewStringArrayVals(v []string) *RecordVals { return &RecordVals{Type: DataTypeStringArray, rawData: encoding.StringArrayToBytes(v)} } -// NewDocVals aliases payload — no copy. The payload must stay unmodified until -// the val is decoded (decoding is lazy: AsDoc may run in a downstream executor -// long after this call) — so pooled or reused buffers are unsafe here. -func NewDocVals(payload []byte) *RecordVals { - return &RecordVals{Type: DataTypeDocument, rawData: payload} +// NewJSONVals aliases payload — no copy. The payload must stay unmodified for +// as long as the val is alive — pooled or reused buffers are unsafe here. +func NewJSONVals(payload []byte) *RecordVals { + return &RecordVals{Type: DataTypeJSON, rawData: payload} } func (rv *RecordVals) RawData() []byte { @@ -167,21 +156,10 @@ func (rv *RecordVals) AsSeqID() seq.ID { return encoding.SeqIDFromBytes(rv.rawData) } -func (rv *RecordVals) AsDoc() *insaneJSON.Root { - rv.mustBe(DataTypeDocument) - if !rv.decoded { - root := insaneJSON.Spawn() - err := root.DecodeBytes(rv.rawData) - if err != nil { - panic(fmt.Errorf("error decoding document: %w", err)) - } - if !root.IsObject() { - panic(fmt.Errorf("document is not an object: %s", rv.rawData)) - } - rv.doc = root - rv.decoded = true - } - return rv.doc +// AsJSON returns the raw json bytes without decoding. +func (rv *RecordVals) AsJSON() []byte { + rv.mustBe(DataTypeJSON) + return rv.rawData } func (rv *RecordVals) AsString() string { @@ -231,16 +209,3 @@ func (rv *RecordVals) AsStringArray() []string { } return rv.strs } - -// Release returns the insaneJSON root allocated in AsDoc back to the library's -// internal pool and invalidates the cache. It is idempotent: after -// the first call the cache is cleared, so repeated calls are a no-op. Calling -// it on a non-document val or a not-yet-decoded val is also a no-op. Safe to -// invoke from every executor that has touched the val — the first caller wins. -func (rv *RecordVals) Release() { - if rv.decoded { - insaneJSON.Release(rv.doc) - } - rv.doc = nil - rv.decoded = false -} diff --git a/query/schema.go b/query/schema.go index d8c67c00..58c898e1 100644 --- a/query/schema.go +++ b/query/schema.go @@ -122,7 +122,7 @@ var ( DataTypeSeqID: (*RecordVals).AsSeqID, DataTypeBytes: (*RecordVals).AsBytes, DataTypeString: (*RecordVals).AsString, - DataTypeDocument: (*RecordVals).AsDoc, + DataTypeJSON: (*RecordVals).AsJSON, DataTypeUint32: (*RecordVals).AsUint32, DataTypeUint64: (*RecordVals).AsUint64, DataTypeInt32: (*RecordVals).AsInt32, @@ -150,7 +150,7 @@ var ( DocsSchema = MustNewSchema( ColumnDesc{Name: DocsIDCol, Type: DataTypeSeqID}, - ColumnDesc{Name: DocsDataCol, Type: DataTypeDocument}, + ColumnDesc{Name: DocsDataCol, Type: DataTypeJSON}, ) ) diff --git a/query/schema_test.go b/query/schema_test.go index 2b918321..9d88cfc0 100644 --- a/query/schema_test.go +++ b/query/schema_test.go @@ -3,7 +3,6 @@ package query import ( "testing" - insaneJSON "github.com/ozontech/insane-json" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -13,7 +12,7 @@ import ( func TestNewSchemaDuplicateName(t *testing.T) { _, err := NewSchema( ColumnDesc{Name: "id", Type: DataTypeSeqID}, - ColumnDesc{Name: "id", Type: DataTypeDocument}, + ColumnDesc{Name: "id", Type: DataTypeJSON}, ) assert.Error(t, err) assert.Equal(t, err.Error(), `duplicate column name: id`) @@ -37,9 +36,9 @@ func TestColumnAllTypes(t *testing.T) { assert.Equal(t, DataTypeString, c.DataType()) }) t.Run("document", func(t *testing.T) { - s := MustNewSchema(ColumnDesc{Name: "d", Type: DataTypeDocument}) - c := s.MustColumn[*insaneJSON.Root]("d") - assert.Equal(t, DataTypeDocument, c.DataType()) + s := MustNewSchema(ColumnDesc{Name: "d", Type: DataTypeJSON}) + c := s.MustColumn[[]byte]("d") + assert.Equal(t, DataTypeJSON, c.DataType()) }) t.Run("numeric", func(t *testing.T) { s := MustNewSchema( @@ -88,7 +87,7 @@ func TestColumnTypeMismatch(t *testing.T) { _, err = DocsSchema.Column[float64]("data") require.Error(t, err) - assert.Equal(t, err.Error(), `column "data" has type document, incompatible with float64`) + assert.Equal(t, err.Error(), `column "data" has type json, incompatible with float64`) } func TestMustColumnPanics(t *testing.T) { @@ -103,14 +102,14 @@ func TestMustColumnPanics(t *testing.T) { func TestSchemaEqual(t *testing.T) { base := MustNewSchema( ColumnDesc{Name: "id", Type: DataTypeSeqID}, - ColumnDesc{Name: "data", Type: DataTypeDocument}, + ColumnDesc{Name: "data", Type: DataTypeJSON}, ) same := MustNewSchema( ColumnDesc{Name: "id", Type: DataTypeSeqID}, - ColumnDesc{Name: "data", Type: DataTypeDocument}, + ColumnDesc{Name: "data", Type: DataTypeJSON}, ) reordered := MustNewSchema( - ColumnDesc{Name: "data", Type: DataTypeDocument}, + ColumnDesc{Name: "data", Type: DataTypeJSON}, ColumnDesc{Name: "id", Type: DataTypeSeqID}, ) retyped := MustNewSchema(