Merge pull request #1589 from ClickHouse/json_serialization_version_2

SpencerTorres · web-flow · commit b8da9f579b7b · 2025-07-08T11:06:28.000-04:00
`JSON` serialization version 3
diff --git a/examples/clickhouse_api/dynamic.go b/examples/clickhouse_api/dynamic.go
@@ -27,13 +27,14 @@ func DynamicExample() error {
 	ctx := context.Background()
 
 	conn, err := GetNativeConnection(clickhouse.Settings{
-		"allow_experimental_dynamic_type": true,
+		"allow_experimental_dynamic_type":                                   true,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, nil)
 	if err != nil {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 8, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for Dynamic type")
 		return nil
 	}
diff --git a/examples/clickhouse_api/json_fast_structs.go b/examples/clickhouse_api/json_fast_structs.go
@@ -91,13 +91,14 @@ func JSONFastStructExample() error {
 	ctx := context.Background()
 
 	conn, err := GetNativeConnection(clickhouse.Settings{
-		"allow_experimental_json_type": true,
+		"allow_experimental_json_type":                                      true,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, nil)
 	if err != nil {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 9, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for JSON type")
 		return nil
 	}
diff --git a/examples/clickhouse_api/json_paths.go b/examples/clickhouse_api/json_paths.go
@@ -28,13 +28,14 @@ func JSONPathsExample() error {
 	ctx := context.Background()
 
 	conn, err := GetNativeConnection(clickhouse.Settings{
-		"allow_experimental_json_type": true,
+		"allow_experimental_json_type":                                      true,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, nil)
 	if err != nil {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 9, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for JSON type")
 		return nil
 	}
diff --git a/examples/clickhouse_api/json_strings.go b/examples/clickhouse_api/json_strings.go
@@ -27,14 +27,15 @@ func JSONStringExample() error {
 	ctx := context.Background()
 
 	conn, err := GetNativeConnection(clickhouse.Settings{
-		"allow_experimental_json_type":              true,
-		"output_format_native_write_json_as_string": true,
+		"allow_experimental_json_type":                                      true,
+		"output_format_native_write_json_as_string":                         true,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, nil)
 	if err != nil {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 9, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for JSON type")
 		return nil
 	}
diff --git a/examples/clickhouse_api/json_structs.go b/examples/clickhouse_api/json_structs.go
@@ -61,13 +61,14 @@ func JSONStructExample() error {
 	ctx := context.Background()
 
 	conn, err := GetNativeConnection(clickhouse.Settings{
-		"allow_experimental_json_type": true,
+		"allow_experimental_json_type":                                      true,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, nil)
 	if err != nil {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 9, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for JSON type")
 		return nil
 	}
diff --git a/examples/clickhouse_api/main_test.go b/examples/clickhouse_api/main_test.go
@@ -234,6 +234,5 @@ func TestJSONFastStructExample(t *testing.T) {
 
 func TestJSONStringExample(t *testing.T) {
 	clickhouse_tests.SkipOnCloud(t, "cannot modify JSON settings on cloud")
-	t.Skip("client cannot receive JSON strings")
 	require.NoError(t, JSONStringExample())
 }
diff --git a/examples/std/json_paths.go b/examples/std/json_paths.go
@@ -32,7 +32,7 @@ func JSONPathsExample() error {
 		return err
 	}
 
-	if !CheckMinServerVersion(conn, 24, 9, 0) {
+	if !CheckMinServerVersion(conn, 25, 6, 0) {
 		fmt.Print("unsupported clickhouse version for JSON type")
 		return nil
 	}
@@ -41,6 +41,10 @@ func JSONPathsExample() error {
 	if err != nil {
 		return err
 	}
+	_, err = conn.ExecContext(ctx, "SET output_format_native_use_flattened_dynamic_and_json_serialization = 1")
+	if err != nil {
+		return err
+	}
 
 	defer func() {
 		conn.Exec("DROP TABLE go_json_example")
diff --git a/examples/std/main_test.go b/examples/std/main_test.go
@@ -166,6 +166,5 @@ func TestJSONPathsExample(t *testing.T) {
 
 func TestJSONStringExample(t *testing.T) {
 	clickhouse_tests.SkipOnCloud(t, "cannot modify JSON settings on cloud")
-	t.Skip("client cannot receive JSON strings")
 	require.NoError(t, JSONStringExample())
 }
diff --git a/lib/column/dynamic.go b/lib/column/dynamic.go
@@ -321,7 +321,7 @@ func (c *Dynamic) decodeHeader(reader *proto.Reader) error {
 	}
 
 	if dynamicSerializationVersion == DeprecatedSupportedDynamicSerializationVersion {
-		return fmt.Errorf("deprecated dynamic serialization version: %d, enable \"output_format_native_use_flattened_dynamic_and_json_serialization\" in your settings.", dynamicSerializationVersion)
+		return fmt.Errorf("deprecated dynamic serialization version: %d, enable \"output_format_native_use_flattened_dynamic_and_json_serialization\" in your settings", dynamicSerializationVersion)
 	} else if dynamicSerializationVersion != SupportedDynamicSerializationVersion {
 		return fmt.Errorf("unsupported dynamic serialization version: %d", dynamicSerializationVersion)
 	}
diff --git a/lib/column/json.go b/lib/column/json.go
@@ -29,8 +29,9 @@ import (
 	"github.com/ClickHouse/ch-go/proto"
 )
 
-const JSONObjectSerializationVersion uint64 = 0
+const JSONDeprecatedObjectSerializationVersion uint64 = 0
 const JSONStringSerializationVersion uint64 = 1
+const JSONObjectSerializationVersion uint64 = 3
 const JSONUnsetSerializationVersion uint64 = math.MaxUint64
 const DefaultMaxDynamicPaths = 1024
 
@@ -51,13 +52,13 @@ type JSON struct {
 	skipPaths      []string
 	skipPathsIndex map[string]int
 
+	totalDynamicPaths int
 	dynamicPaths      []string
 	dynamicPathsIndex map[string]int
 	dynamicColumns    []*Dynamic
 
-	maxDynamicPaths   int
-	maxDynamicTypes   int
-	totalDynamicPaths int
+	maxDynamicPaths int
+	maxDynamicTypes int
 }
 
 func (c *JSON) parse(t Type, tz *time.Location) (_ *JSON, err error) {
@@ -544,17 +545,29 @@ func (c *JSON) appendRowString(v any) error {
 	return nil
 }
 
-func (c *JSON) encodeObjectHeader(buffer *proto.Buffer) {
-	buffer.PutUVarInt(uint64(c.maxDynamicPaths))
+func (c *JSON) encodeObjectHeader(buffer *proto.Buffer) error {
 	buffer.PutUVarInt(uint64(c.totalDynamicPaths))
 
 	for _, dynamicPath := range c.dynamicPaths {
 		buffer.PutString(dynamicPath)
 	}
 
-	for _, col := range c.dynamicColumns {
-		col.encodeHeader(buffer)
+	for i, col := range c.typedColumns {
+		if serialize, ok := col.(CustomSerialization); ok {
+			if err := serialize.WriteStatePrefix(buffer); err != nil {
+				return fmt.Errorf("failed to write prefix for typed path \"%s\" in json with type %s: %w", c.typedPaths[i], string(col.Type()), err)
+			}
+		}
+	}
+
+	for i, col := range c.dynamicColumns {
+		err := col.encodeHeader(buffer)
+		if err != nil {
+			return fmt.Errorf("failed to encode header for json dynamic path \"%s\": %w", c.dynamicPaths[i], err)
+		}
 	}
+
+	return nil
 }
 
 func (c *JSON) encodeObjectData(buffer *proto.Buffer) {
@@ -565,11 +578,6 @@ func (c *JSON) encodeObjectData(buffer *proto.Buffer) {
 	for _, col := range c.dynamicColumns {
 		col.encodeData(buffer)
 	}
-
-	// SharedData per row, empty for now.
-	for i := 0; i < c.rows; i++ {
-		buffer.PutUInt64(0)
-	}
 }
 
 func (c *JSON) encodeStringData(buffer *proto.Buffer) {
@@ -580,9 +588,7 @@ func (c *JSON) WriteStatePrefix(buffer *proto.Buffer) error {
 	switch c.serializationVersion {
 	case JSONObjectSerializationVersion:
 		buffer.PutUInt64(JSONObjectSerializationVersion)
-		c.encodeObjectHeader(buffer)
-
-		return nil
+		return c.encodeObjectHeader(buffer)
 	case JSONStringSerializationVersion:
 		buffer.PutUInt64(JSONStringSerializationVersion)
 
@@ -632,20 +638,14 @@ func (c *JSON) Reset() {
 }
 
 func (c *JSON) decodeObjectHeader(reader *proto.Reader) error {
-	maxDynamicPaths, err := reader.UVarInt()
-	if err != nil {
-		return fmt.Errorf("failed to read max dynamic paths for json column: %w", err)
-	}
-	c.maxDynamicPaths = int(maxDynamicPaths)
-
 	totalDynamicPaths, err := reader.UVarInt()
 	if err != nil {
 		return fmt.Errorf("failed to read total dynamic paths for json column: %w", err)
 	}
 	c.totalDynamicPaths = int(totalDynamicPaths)
 
-	c.dynamicPaths = make([]string, 0, totalDynamicPaths)
-	for i := 0; i < int(totalDynamicPaths); i++ {
+	c.dynamicPaths = make([]string, 0, c.totalDynamicPaths)
+	for i := 0; i < c.totalDynamicPaths; i++ {
 		dynamicPath, err := reader.Str()
 		if err != nil {
 			return fmt.Errorf("failed to read dynamic path name bytes at index %d for json column: %w", i, err)
@@ -655,14 +655,22 @@ func (c *JSON) decodeObjectHeader(reader *proto.Reader) error {
 		c.dynamicPathsIndex[dynamicPath] = len(c.dynamicPaths) - 1
 	}
 
-	c.dynamicColumns = make([]*Dynamic, 0, totalDynamicPaths)
+	for i, col := range c.typedColumns {
+		if serialize, ok := col.(CustomSerialization); ok {
+			if err := serialize.ReadStatePrefix(reader); err != nil {
+				return fmt.Errorf("failed to read prefix for typed path \"%s\" with type %s in json: %w", c.typedPaths[i], string(col.Type()), err)
+			}
+		}
+	}
+
+	c.dynamicColumns = make([]*Dynamic, 0, c.totalDynamicPaths)
 	for _, dynamicPath := range c.dynamicPaths {
 		parsedColDynamic, _ := Type("Dynamic").Column("", c.tz)
 		colDynamic := parsedColDynamic.(*Dynamic)
 
 		err := colDynamic.decodeHeader(reader)
 		if err != nil {
-			return fmt.Errorf("failed to decode dynamic header at path %s for json column: %w", dynamicPath, err)
+			return fmt.Errorf("failed to decode dynamic header at path \"%s\" for json column: %w", dynamicPath, err)
 		}
 
 		c.dynamicColumns = append(c.dynamicColumns, colDynamic)
@@ -677,7 +685,7 @@ func (c *JSON) decodeObjectData(reader *proto.Reader, rows int) error {
 
 		err := col.Decode(reader, rows)
 		if err != nil {
-			return fmt.Errorf("failed to decode %s typed path %s for json column: %w", col.Type(), typedPath, err)
+			return fmt.Errorf("failed to decode %s typed path \"%s\" for json column: %w", col.Type(), typedPath, err)
 		}
 	}
 
@@ -686,16 +694,10 @@ func (c *JSON) decodeObjectData(reader *proto.Reader, rows int) error {
 
 		err := col.decodeData(reader, rows)
 		if err != nil {
-			return fmt.Errorf("failed to decode dynamic path %s for json column: %w", dynamicPath, err)
+			return fmt.Errorf("failed to decode dynamic path \"%s\" for json column: %w", dynamicPath, err)
 		}
 	}
 
-	// SharedData per row, ignored for now. May cause stream offset issues if present
-	_, err := reader.ReadRaw(8 * rows) // one UInt64 per row
-	if err != nil {
-		return fmt.Errorf("failed to read shared data for json column: %w", err)
-	}
-
 	return nil
 }
 
@@ -712,6 +714,8 @@ func (c *JSON) ReadStatePrefix(reader *proto.Reader) error {
 	c.serializationVersion = jsonSerializationVersion
 
 	switch jsonSerializationVersion {
+	case JSONDeprecatedObjectSerializationVersion:
+		return fmt.Errorf("deprecated json serialization version: %d, enable \"output_format_native_use_flattened_dynamic_and_json_serialization\" in your settings", jsonSerializationVersion)
 	case JSONObjectSerializationVersion:
 		err := c.decodeObjectHeader(reader)
 		if err != nil {
diff --git a/tests/dynamic_test.go b/tests/dynamic_test.go
@@ -36,7 +36,7 @@ func setupDynamicTest(t *testing.T, protocol clickhouse.Protocol) driver.Conn {
 	conn, err := GetNativeConnection(t, protocol, clickhouse.Settings{
 		"max_execution_time":              60,
 		"allow_experimental_dynamic_type": true,
-		"output_format_native_use_flattened_dynamic_and_json_serialization": 1,
+		"output_format_native_use_flattened_dynamic_and_json_serialization": true,
 	}, nil, &clickhouse.Compression{
 		Method: clickhouse.CompressionLZ4,
 	})
diff --git a/tests/json_test.go b/tests/json_test.go
diff --git a/tests/std/dynamic_test.go b/tests/std/dynamic_test.go
diff --git a/tests/std/json_test.go b/tests/std/json_test.go

Original file line number	Diff line number	Diff line change
`@@ -234,6 +234,5 @@ func TestJSONFastStructExample(t *testing.T) {`
`234`	`234`
`235`	`235`	`func TestJSONStringExample(t *testing.T) {`
`236`	`236`	`clickhouse_tests.SkipOnCloud(t, "cannot modify JSON settings on cloud")`
`237`		`- t.Skip("client cannot receive JSON strings")`
`238`	`237`	`require.NoError(t, JSONStringExample())`
`239`	`238`	`}`
Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ func JSONPathsExample() error {`
`32`	`32`	`return err`
`33`	`33`	`}`
`34`	`34`
`35`		`- if !CheckMinServerVersion(conn, 24, 9, 0) {`
	`35`	`+ if !CheckMinServerVersion(conn, 25, 6, 0) {`
`36`	`36`	`fmt.Print("unsupported clickhouse version for JSON type")`
`37`	`37`	`return nil`
`38`	`38`	`}`
`@@ -41,6 +41,10 @@ func JSONPathsExample() error {`
`41`	`41`	`if err != nil {`
`42`	`42`	`return err`
`43`	`43`	`}`
	`44`	`+ _, err = conn.ExecContext(ctx, "SET output_format_native_use_flattened_dynamic_and_json_serialization = 1")`
	`45`	`+ if err != nil {`
	`46`	`+ return err`
	`47`	`+ }`
`44`	`48`
`45`	`49`	`defer func() {`
`46`	`50`	`conn.Exec("DROP TABLE go_json_example")`
Original file line number	Diff line number	Diff line change
`@@ -166,6 +166,5 @@ func TestJSONPathsExample(t *testing.T) {`
`166`	`166`
`167`	`167`	`func TestJSONStringExample(t *testing.T) {`
`168`	`168`	`clickhouse_tests.SkipOnCloud(t, "cannot modify JSON settings on cloud")`
`169`		`- t.Skip("client cannot receive JSON strings")`
`170`	`169`	`require.NoError(t, JSONStringExample())`
`171`	`170`	`}`
Original file line number	Diff line number	Diff line change
`@@ -321,7 +321,7 @@ func (c Dynamic) decodeHeader(reader proto.Reader) error {`
`321`	`321`	`}`
`322`	`322`
`323`	`323`	`if dynamicSerializationVersion == DeprecatedSupportedDynamicSerializationVersion {`
`324`		`- return fmt.Errorf("deprecated dynamic serialization version: %d, enable \"output_format_native_use_flattened_dynamic_and_json_serialization\" in your settings.", dynamicSerializationVersion)`
	`324`	`+ return fmt.Errorf("deprecated dynamic serialization version: %d, enable \"output_format_native_use_flattened_dynamic_and_json_serialization\" in your settings", dynamicSerializationVersion)`
`325`	`325`	`} else if dynamicSerializationVersion != SupportedDynamicSerializationVersion {`
`326`	`326`	`return fmt.Errorf("unsupported dynamic serialization version: %d", dynamicSerializationVersion)`
`327`	`327`	`}`