CHASM: path encoder implementation (#7891)

## What changed?
- Implement CHASM path encoder

## Why?
- CHASM work stream

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

chasm/field_test.go

@@ -114,7 +114,7 @@ func (s *fieldSuite) TestFieldGetComponent() {
 
 	chasmContext := NewMutableContext(context.Background(), node)
 
-	c, err := node.Component(chasmContext, ComponentRef{componentPath: RootPath})
+	c, err := node.Component(chasmContext, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	s.NotNil(c)
 

chasm/path_encoder.go

@@ -1,6 +1,11 @@
 package chasm
 
-import "strings"
+import (
+	"strings"
+	"unicode/utf8"
+
+	"go.temporal.io/api/serviceerror"
+)
 
 var _ NodePathEncoder = (*defaultPathEncoder)(nil)
 
@@ -8,22 +13,120 @@ var DefaultPathEncoder NodePathEncoder = &defaultPathEncoder{}
 
 type defaultPathEncoder struct{}
 
-// TODO: Have a real implementation for DefaultPathEncoder
-// that handles special characters in the path and support
-// getting all immedidate children of a collection node.
+const (
+	nameSeparator       = '$'
+	collectionSeparator = '#'
+	escapeChar          = '\\'
+)
 
+var (
+	rootPath = []string{}
+)
+
+// The Encode method encodes node path in a way that the following uses cases can be
+// achieved by doing a simple a range query in DB based on prefixes of the encoded path:
+// 1. Getting all nodes for a chasm tree.
+// 2. Getting all nodes for a sub-tree.
+// 3. Getting all immediate children of a Collection node.
+// Additionally, it allows getting all ancestor nodes of a given node.
+//
+// It does so by using a different separator for a node which is a direct child of a Collection node.
+// The two separators used ('$' and '#') are next to each other in terms of values, which ensures all
+// children for a node are grouped together. Additionally, all immediate children of a Collection node
+// are grouped together as well.
+//
+// To get a sub-tree (say a node with name "foo"), we want to use a query look like the following:
+//
+//	path >= "foo" AND path < "foo[something]"
+//
+// and we need to know the minimal value of [something] that can possibly be in the encoded path.
+// This is achieved by escaping '\', '$', '#', and also every code point less than '#'.
+// Since the escaped character itself is larger than '#', the minimal value is '%' and our query becomes:
+//
+//	path >= "foo" AND path < "foo%"
 func (e *defaultPathEncoder) Encode(
-	_ *Node,
+	node *Node,
 	path []string,
 ) (string, error) {
-	return strings.Join(path, "/"), nil
+	if path == nil {
+		path = node.path()
+	}
+
+	if len(path) == 0 {
+		return "", nil
+	}
+
+	var b strings.Builder
+	lastIdx := len(path) - 1
+	for i, nodeName := range path {
+		if i > 0 {
+			if i == lastIdx &&
+				node.parent != nil &&
+				node.parent.serializedNode.GetMetadata().GetCollectionAttributes() != nil {
+				_, _ = b.WriteRune(collectionSeparator)
+			} else {
+				_, _ = b.WriteRune(nameSeparator)
+			}
+		}
+
+		if nodeName == "" {
+			return "", serviceerror.NewInternalf("path contains empty node name: %v", path)
+		}
+
+		for _, r := range nodeName {
+			if r == utf8.RuneError {
+				return "", serviceerror.NewInvalidArgumentf("node name contains invalid UTF-8 code point: %v", nodeName)
+			}
+
+			if r == escapeChar ||
+				r == nameSeparator ||
+				r <= collectionSeparator {
+				_, _ = b.WriteRune(escapeChar)
+			}
+			_, _ = b.WriteRune(r)
+		}
+	}
+	return b.String(), nil
 }
 
 func (e *defaultPathEncoder) Decode(
 	encodedPath string,
 ) ([]string, error) {
 	if encodedPath == "" {
-		return []string{}, nil
+		return rootPath, nil
 	}
-	return strings.Split(encodedPath, "/"), nil
+
+	path := make([]string, 0, 3)
+	var b strings.Builder
+	escaped := false
+	for _, r := range encodedPath {
+		if r == utf8.RuneError {
+			return nil, serviceerror.NewInvalidArgumentf("encodedPath contains invalid UTF-8 code point: %v", encodedPath)
+		}
+
+		if escaped {
+			_, _ = b.WriteRune(r)
+			escaped = false
+			continue
+		}
+
+		if r == '\\' {
+			escaped = true
+			continue
+		}
+
+		if r == '$' || r == '#' {
+			path = append(path, b.String())
+			b.Reset()
+			continue
+		}
+
+		_, _ = b.WriteRune(r)
+	}
+	if escaped {
+		return nil, serviceerror.NewInternalf("encoded path ends with escape character: %v", encodedPath)
+	}
+
+	path = append(path, b.String())
+	return path, nil
 }

chasm/path_encoder_test.go

@@ -0,0 +1,88 @@
+package chasm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+)
+
+func TestDefaultPathEncoder_EncodeDecode(t *testing.T) {
+	e := &defaultPathEncoder{}
+
+	root := &Node{
+		nodeName: "",
+		serializedNode: &persistencespb.ChasmNode{
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{ComponentAttributes: &persistencespb.ChasmComponentAttributes{}},
+			},
+		},
+	}
+	child := &Node{
+		parent:   root,
+		nodeName: "child",
+		serializedNode: &persistencespb.ChasmNode{
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{ComponentAttributes: &persistencespb.ChasmComponentAttributes{}},
+			},
+		},
+	}
+	collection := &Node{
+		parent:   root,
+		nodeName: "collection",
+		serializedNode: &persistencespb.ChasmNode{
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				Attributes: &persistencespb.ChasmNodeMetadata_CollectionAttributes{CollectionAttributes: &persistencespb.ChasmCollectionAttributes{}},
+			},
+		},
+	}
+	collectionItem := &Node{
+		parent:   collection,
+		nodeName: "item",
+		serializedNode: &persistencespb.ChasmNode{
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{ComponentAttributes: &persistencespb.ChasmComponentAttributes{}},
+			},
+		},
+	}
+	collectionItemData := &Node{
+		parent:   collectionItem,
+		nodeName: "data",
+		serializedNode: &persistencespb.ChasmNode{
+			Metadata: &persistencespb.ChasmNodeMetadata{
+				Attributes: &persistencespb.ChasmNodeMetadata_DataAttributes{},
+			},
+		},
+	}
+
+	tests := []struct {
+		node    *Node
+		path    []string
+		encoded string
+	}{
+		{root, []string{}, ""},
+
+		{child, []string{"child"}, "child"},
+		{child, []string{"special\\#"}, "special\\\\\\#"},
+		{child, []string{" !"}, "\\ \\!"},
+		{child, []string{"你好"}, "你好"},
+
+		{collection, []string{"collection"}, "collection"},
+
+		{collectionItem, []string{"collection", "item"}, "collection#item"},
+		{collectionItem, []string{"collection", "⌘"}, "collection#⌘"},
+
+		{collectionItemData, []string{"collection", "item", "data"}, "collection$item$data"},
+		{collectionItemData, []string{"collection", "item", "世界"}, "collection$item$世界"},
+	}
+
+	for _, tt := range tests {
+		encoded, err := e.Encode(tt.node, tt.path)
+		require.NoError(t, err)
+		require.Equal(t, tt.encoded, encoded)
+
+		decodedPath, err := e.Decode(encoded)
+		require.NoError(t, err)
+		require.Equal(t, tt.path, decodedPath)
+	}
+}

chasm/ref.go

@@ -10,7 +10,6 @@ import (
 
 var (
 	defaultShardingFn = func(key EntityKey) string { return key.NamespaceID + "_" + key.BusinessID }
-	RootPath          []string
 )
 
 type EntityKey struct {

chasm/tree.go

@@ -63,7 +63,7 @@ type (
 
 		parent   *Node
 		children map[string]*Node // child name (path segment) -> child node
-		nodeName string           // key of this node in parent's children map.
+		nodeName string           // key of this node in parent's children map, empty string for root node.
 
 		// Type of attributes controls the type of the node.
 		serializedNode *persistencespb.ChasmNode // serialized component | data | collection with metadata
@@ -568,7 +568,7 @@ func (n *Node) syncSubComponents() error {
 	if n.value == nil {
 		return nil
 	}
-	return n.syncSubComponentsInternal(RootPath)
+	return n.syncSubComponentsInternal(rootPath)
 }
 
 func (n *Node) syncSubComponentsInternal(
@@ -1593,7 +1593,7 @@ func (n *Node) encodedPath() (string, error) {
 
 func (n *Node) path() []string {
 	if n.parent == nil {
-		return []string{n.nodeName}
+		return []string{}
 	}
 
 	return append(n.parent.path(), n.nodeName)

chasm/tree_test.go

@@ -1305,7 +1305,7 @@ func (s *nodeSuite) TestCloseTransaction_Success() {
 	tv := testvars.New(s.T())
 
 	chasmCtx := NewMutableContext(context.Background(), node)
-	tc, err := node.Component(chasmCtx, ComponentRef{componentPath: RootPath})
+	tc, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	tc.(*TestComponent).SubData1 = NewEmptyField[*protoMessageType]()
 	tc.(*TestComponent).ComponentData = &protoMessageType{CreateRequestId: tv.Any().String()}
@@ -1363,7 +1363,7 @@ func (s *nodeSuite) TestCloseTransaction_LifecycleChange() {
 	tv := testvars.New(s.T())
 
 	chasmCtx := NewMutableContext(context.Background(), node)
-	_, err := node.Component(chasmCtx, ComponentRef{componentPath: RootPath})
+	_, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 
 	s.nodeBackend.EXPECT().NextTransitionCount().Return(int64(1)).AnyTimes()
@@ -1378,7 +1378,7 @@ func (s *nodeSuite) TestCloseTransaction_LifecycleChange() {
 	s.NoError(err)
 
 	// Test force terminate case
-	_, err = node.Component(chasmCtx, ComponentRef{componentPath: RootPath})
+	_, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	node.terminated = true
 	s.nodeBackend.EXPECT().UpdateWorkflowStateStatus(
@@ -1389,7 +1389,7 @@ func (s *nodeSuite) TestCloseTransaction_LifecycleChange() {
 	s.NoError(err)
 	node.terminated = false
 
-	tc, err := node.Component(chasmCtx, ComponentRef{componentPath: RootPath})
+	tc, err := node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	tc.(*TestComponent).Complete(chasmCtx)
 	s.nodeBackend.EXPECT().UpdateWorkflowStateStatus(
@@ -1399,7 +1399,7 @@ func (s *nodeSuite) TestCloseTransaction_LifecycleChange() {
 	_, err = node.CloseTransaction()
 	s.NoError(err)
 
-	tc, err = node.Component(chasmCtx, ComponentRef{componentPath: RootPath})
+	tc, err = node.Component(chasmCtx, ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	tc.(*TestComponent).Fail(chasmCtx)
 	s.nodeBackend.EXPECT().UpdateWorkflowStateStatus(
@@ -1826,7 +1826,7 @@ func (e *testNodePathEncoder) Decode(
 	encodedPath string,
 ) ([]string, error) {
 	if encodedPath == "" {
-		return []string{}, nil
+		return rootPath, nil
 	}
 	return strings.Split(encodedPath, "/"), nil
 }
@@ -1858,7 +1858,7 @@ func (s *nodeSuite) testComponentTree() *Node {
 	s.IsType(&TestComponent{}, node.value)
 	s.Equal(valueStateSynced, node.valueState)
 
-	tc, err := node.Component(NewMutableContext(context.Background(), node), ComponentRef{componentPath: RootPath})
+	tc, err := node.Component(NewMutableContext(context.Background(), node), ComponentRef{componentPath: rootPath})
 	s.NoError(err)
 	s.Equal(valueStateNeedSerialize, node.valueState)
 	// Create subcomponents by assigning fields to TestComponent instance.