Merge branch 'main' into feat/otlp-resource-descriptor

This commit is contained in:
dennis zhuang
2026-08-22 22:05:33 +08:00
committed by GitHub
45 changed files with 3331 additions and 432 deletions
Generated
+84 -52
View File
@@ -977,12 +977,6 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "auto_ops"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7460f7dd8e100147b82a63afca1a20eb6c231ee36b90ba7272e14951cb58af59"
[[package]]
name = "autocfg"
version = "1.5.0"
@@ -3744,7 +3738,7 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
[[package]]
name = "datafusion"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"arrow-schema 58.3.0",
@@ -3798,7 +3792,7 @@ dependencies = [
[[package]]
name = "datafusion-catalog"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -3822,7 +3816,7 @@ dependencies = [
[[package]]
name = "datafusion-catalog-listing"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -3844,7 +3838,7 @@ dependencies = [
[[package]]
name = "datafusion-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -3868,7 +3862,7 @@ dependencies = [
[[package]]
name = "datafusion-common-runtime"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"futures",
"log",
@@ -3878,7 +3872,7 @@ dependencies = [
[[package]]
name = "datafusion-datasource"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-compression",
@@ -3912,7 +3906,7 @@ dependencies = [
[[package]]
name = "datafusion-datasource-arrow"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"arrow-ipc 58.3.0",
@@ -3935,7 +3929,7 @@ dependencies = [
[[package]]
name = "datafusion-datasource-csv"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -3957,7 +3951,7 @@ dependencies = [
[[package]]
name = "datafusion-datasource-json"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -3980,7 +3974,7 @@ dependencies = [
[[package]]
name = "datafusion-datasource-parquet"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -4009,12 +4003,12 @@ dependencies = [
[[package]]
name = "datafusion-doc"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
[[package]]
name = "datafusion-execution"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"arrow-buffer 58.3.0",
@@ -4036,7 +4030,7 @@ dependencies = [
[[package]]
name = "datafusion-expr"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -4058,7 +4052,7 @@ dependencies = [
[[package]]
name = "datafusion-expr-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4070,7 +4064,7 @@ dependencies = [
[[package]]
name = "datafusion-functions"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"arrow-buffer 58.3.0",
@@ -4101,7 +4095,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-aggregate"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -4122,7 +4116,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-aggregate-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -4134,7 +4128,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-nested"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"arrow-ord 58.3.0",
@@ -4158,7 +4152,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-table"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"async-trait",
@@ -4173,7 +4167,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-window"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4190,7 +4184,7 @@ dependencies = [
[[package]]
name = "datafusion-functions-window-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"datafusion-common",
"datafusion-physical-expr-common",
@@ -4199,7 +4193,7 @@ dependencies = [
[[package]]
name = "datafusion-macros"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"datafusion-doc",
"quote",
@@ -4209,7 +4203,7 @@ dependencies = [
[[package]]
name = "datafusion-optimizer"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"chrono",
@@ -4258,7 +4252,7 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -4281,7 +4275,7 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr-adapter"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4295,7 +4289,7 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -4311,7 +4305,7 @@ dependencies = [
[[package]]
name = "datafusion-physical-optimizer"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4329,7 +4323,7 @@ dependencies = [
[[package]]
name = "datafusion-physical-plan"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"ahash 0.8.12",
"arrow 58.3.0",
@@ -4360,7 +4354,7 @@ dependencies = [
[[package]]
name = "datafusion-proto"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"chrono",
@@ -4387,7 +4381,7 @@ dependencies = [
[[package]]
name = "datafusion-proto-common"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4397,7 +4391,7 @@ dependencies = [
[[package]]
name = "datafusion-pruning"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"datafusion-common",
@@ -4413,7 +4407,7 @@ dependencies = [
[[package]]
name = "datafusion-session"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"async-trait",
"datafusion-common",
@@ -4426,7 +4420,7 @@ dependencies = [
[[package]]
name = "datafusion-sql"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"arrow 58.3.0",
"bigdecimal 0.4.8",
@@ -4444,7 +4438,7 @@ dependencies = [
[[package]]
name = "datafusion-substrait"
version = "53.1.0"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=d705d4433d2fcbe2bead54e3f1d6cc59727857e1#d705d4433d2fcbe2bead54e3f1d6cc59727857e1"
source = "git+https://github.com/GreptimeTeam/datafusion.git?rev=4c8a6bf28347b346348136247eb67dfb933f1147#4c8a6bf28347b346348136247eb67dfb933f1147"
dependencies = [
"async-recursion",
"async-trait",
@@ -4554,6 +4548,9 @@ dependencies = [
"num",
"num-traits",
"ordered-float 4.6.0",
"parquet-variant",
"parquet-variant-compute",
"parquet-variant-json",
"paste",
"regex",
"serde",
@@ -9799,17 +9796,6 @@ dependencies = [
"url",
]
[[package]]
name = "openmetrics-parser"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e40a68c62e09c5dfec2f6472af3bd5e8ddf506fcf14c78ece23794ffbb874eca"
dependencies = [
"auto_ops",
"pest",
"pest_derive",
]
[[package]]
name = "opensrv-mysql"
version = "0.8.0"
@@ -10259,6 +10245,53 @@ dependencies = [
"zstd",
]
[[package]]
name = "parquet-variant"
version = "58.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "74c8db065291f088a2aad8ab831853eae1871c0d311c8d0b83bbc3b7e735d0fc"
dependencies = [
"arrow 58.3.0",
"arrow-schema 58.3.0",
"chrono",
"half",
"indexmap 2.13.0",
"num-traits",
"simdutf8",
"uuid",
]
[[package]]
name = "parquet-variant-compute"
version = "58.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a530e8d5b5e14efcb39c9a6ec55432ad11f6afb7dc4455a79be0dc615fe3cc31"
dependencies = [
"arrow 58.3.0",
"arrow-schema 58.3.0",
"chrono",
"half",
"indexmap 2.13.0",
"parquet-variant",
"parquet-variant-json",
"serde_json",
"uuid",
]
[[package]]
name = "parquet-variant-json"
version = "58.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "00ed89908289f67caa2ca078f9ff9aacd6229a313ec92b12bf4f48f613dc2b97"
dependencies = [
"arrow-schema 58.3.0",
"base64 0.22.1",
"chrono",
"parquet-variant",
"serde_json",
"uuid",
]
[[package]]
name = "parse-size"
version = "1.1.0"
@@ -13411,7 +13444,6 @@ dependencies = [
"notify",
"object-pool",
"once_cell",
"openmetrics-parser",
"opensrv-mysql",
"opentelemetry-proto",
"operator",
+18 -15
View File
@@ -195,6 +195,9 @@ otel-arrow-rust = { git = "https://github.com/GreptimeTeam/otel-arrow", rev = "5
] }
parking_lot = "0.12"
parquet = { version = "58.3", default-features = false, features = ["arrow", "async", "object_store"] }
parquet-variant = "58.3"
parquet-variant-compute = "58.3"
parquet-variant-json = "58.3"
paste = "1.0"
pin-project = "1.0"
pretty_assertions = "1.4.0"
@@ -348,21 +351,21 @@ git = "https://github.com/GreptimeTeam/greptime-meter.git"
rev = "5618e779cf2bb4755b499c630fba4c35e91898cb"
[patch.crates-io]
datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "d705d4433d2fcbe2bead54e3f1d6cc59727857e1" }
datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-datasource = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-functions-window-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-physical-expr-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-proto = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "4c8a6bf28347b346348136247eb67dfb933f1147" }
sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "2aefa08a8d69c96eec2d6d6703598a009bba6e4c" } # on branch v0.61.x
[profile.release]
+5 -5
View File
@@ -25,7 +25,7 @@
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -256,7 +256,7 @@
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -400,7 +400,7 @@
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
| `backend_tls` | -- | -- | TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)<br/>When using etcd, PostgreSQL, or MySQL as metadata store, you can configure TLS here<br/><br/>Note: if TLS is configured in both this section and the `store_addrs` connection string, the<br/>settings here will override the TLS settings in `store_addrs`. |
| `backend_tls.mode` | String | `prefer` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- "disable" - No TLS<br/>- "prefer" (default) - Try TLS, fallback to plain<br/>- "require" - Require TLS<br/>- "verify_ca" - Require TLS and verify CA<br/>- "verify_full" - Require TLS and verify hostname |
| `backend_tls.cert_path` | String | `""` | Path to client certificate file (for client authentication)<br/>Like "/path/to/client.crt" |
@@ -511,8 +511,8 @@
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.<br/>Defaults to max(num_cpus - 1, 1). |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
| `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.<br/>Defaults to max(num_cpus - 1, 2). |
| `runtime.ingest_rt_size` | Integer | `8` | The number of threads to execute datanode ingestion operations. |
| `meta_client` | -- | -- | The metasrv client options. |
| `meta_client.metasrv_addrs` | Array | -- | The addresses of the metasrv. |
+2 -2
View File
@@ -84,10 +84,10 @@ watch = false
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
## Defaults to max(num_cpus / 2, 2).
#+ compact_rt_max_blocking_threads = 4
## The number of threads to execute datanode query operations.
## Defaults to max(num_cpus - 1, 1).
## Defaults to max(num_cpus - 1, 2).
#+ query_rt_size = 7
## The number of threads to execute datanode ingestion operations.
#+ ingest_rt_size = 8
+1 -1
View File
@@ -44,7 +44,7 @@ default_column_prefix = "greptime"
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
## Defaults to max(num_cpus / 2, 2).
#+ compact_rt_max_blocking_threads = 4
## The HTTP server options.
+1 -1
View File
@@ -92,7 +92,7 @@ node_max_idle_time = "24hours"
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
## Defaults to max(num_cpus / 2, 2).
#+ compact_rt_max_blocking_threads = 4
## TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)
+1 -1
View File
@@ -57,7 +57,7 @@ max_concurrent_queries = 0
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
## Defaults to max(num_cpus / 2, 2).
#+ compact_rt_max_blocking_threads = 4
## The HTTP server options.
+3 -3
View File
@@ -61,9 +61,6 @@ impl InstanceBuilder {
}
async fn init(opts: &mut DatanodeOptions, plugins: &mut Plugins) -> Result<Vec<WorkerGuard>> {
common_runtime::init_global_runtimes(&opts.runtime);
common_runtime::init_datanode_runtimes(&opts.runtime);
let dn_opts = &mut opts.component;
let guard = common_telemetry::init_global_logging(
APP_NAME,
@@ -73,6 +70,9 @@ impl InstanceBuilder {
None,
);
common_runtime::init_global_runtimes(&opts.runtime);
common_runtime::init_datanode_runtimes(&opts.runtime);
crate::options::flush_dropped_plugin_warnings();
log_versions(verbose_version(), short_version(), APP_NAME);
maybe_activate_heap_profile(&dn_opts.memory);
+2 -2
View File
@@ -248,8 +248,6 @@ impl StartCommand {
}
async fn build(&self, opts: FlownodeOptions) -> Result<Instance> {
common_runtime::init_global_runtimes(&opts.runtime);
let guard = common_telemetry::init_global_logging(
APP_NAME,
&opts.component.logging,
@@ -258,6 +256,8 @@ impl StartCommand {
None,
);
common_runtime::init_global_runtimes(&opts.runtime);
crate::options::flush_dropped_plugin_warnings();
log_versions(verbose_version(), short_version(), APP_NAME);
maybe_activate_heap_profile(&opts.component.memory);
+2 -2
View File
@@ -330,8 +330,6 @@ impl StartCommand {
}
async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
common_runtime::init_global_runtimes(&opts.runtime);
let guard = common_telemetry::init_global_logging(
APP_NAME,
&opts.component.logging,
@@ -340,6 +338,8 @@ impl StartCommand {
Some(&opts.component.slow_query),
);
common_runtime::init_global_runtimes(&opts.runtime);
crate::options::flush_dropped_plugin_warnings();
log_versions(verbose_version(), short_version(), APP_NAME);
maybe_activate_heap_profile(&opts.component.memory);
+2 -2
View File
@@ -318,8 +318,6 @@ impl StartCommand {
}
pub async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
common_runtime::init_global_runtimes(&opts.runtime);
let guard = common_telemetry::init_global_logging(
APP_NAME,
&opts.component.logging,
@@ -328,6 +326,8 @@ impl StartCommand {
None,
);
common_runtime::init_global_runtimes(&opts.runtime);
crate::options::flush_dropped_plugin_warnings();
log_versions(verbose_version(), short_version(), APP_NAME);
maybe_activate_heap_profile(&opts.component.memory);
+2 -2
View File
@@ -395,8 +395,6 @@ impl StartCommand {
#[allow(clippy::diverging_sub_expression)]
/// Build GreptimeDB instance with the loaded options.
pub async fn build(&self, opts: GreptimeOptions<StandaloneOptions>) -> Result<Instance> {
common_runtime::init_global_runtimes(&opts.runtime);
let guard = common_telemetry::init_global_logging(
APP_NAME,
&opts.component.logging,
@@ -405,6 +403,8 @@ impl StartCommand {
Some(&opts.component.slow_query),
);
common_runtime::init_global_runtimes(&opts.runtime);
crate::options::flush_dropped_plugin_warnings();
log_versions(verbose_version(), short_version(), APP_NAME);
maybe_activate_heap_profile(&opts.component.memory);
+57 -10
View File
@@ -27,6 +27,9 @@ use crate::{Builder, JoinHandle, Runtime};
const GLOBAL_WORKERS: usize = 8;
const COMPACT_WORKERS: usize = 4;
const HB_WORKERS: usize = 2;
/// The minimum number of worker threads for runtimes sized by CPU count.
/// A single-threaded runtime can easily deadlock in async code.
const MIN_RUNTIME_THREADS: usize = 2;
/// The options for the global runtimes.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
@@ -44,19 +47,25 @@ pub struct RuntimeOptions {
pub ingest_rt_size: usize,
}
impl Default for RuntimeOptions {
fn default() -> Self {
let cpus = num_cpus::get();
impl RuntimeOptions {
fn with_num_cpus(cpus: usize) -> Self {
let cpus = usize::max(cpus, MIN_RUNTIME_THREADS);
Self {
global_rt_size: cpus,
compact_rt_size: usize::max(cpus / 2, 1),
compact_rt_max_blocking_threads: usize::max(cpus / 2, 1),
query_rt_size: usize::max(cpus.saturating_sub(1), 1),
compact_rt_size: usize::max(cpus / 2, MIN_RUNTIME_THREADS),
compact_rt_max_blocking_threads: usize::max(cpus / 2, MIN_RUNTIME_THREADS),
query_rt_size: usize::max(cpus.saturating_sub(1), MIN_RUNTIME_THREADS),
ingest_rt_size: cpus,
}
}
}
impl Default for RuntimeOptions {
fn default() -> Self {
Self::with_num_cpus(num_cpus::get())
}
}
pub fn create_runtime(runtime_name: &str, thread_name: &str, worker_threads: usize) -> Runtime {
info!(
"Creating runtime with runtime_name: {runtime_name}, thread_name: {thread_name}, work_threads: {worker_threads}."
@@ -287,18 +296,56 @@ mod tests {
#[test]
fn test_datanode_runtime_options_default() {
let options = RuntimeOptions::default();
let cpus = num_cpus::get();
let cpus = usize::max(num_cpus::get(), MIN_RUNTIME_THREADS);
assert_eq!(cpus, options.global_rt_size);
assert_eq!(usize::max(cpus / 2, 1), options.compact_rt_size);
assert_eq!(
usize::max(cpus / 2, 1),
usize::max(cpus / 2, MIN_RUNTIME_THREADS),
options.compact_rt_size
);
assert_eq!(
usize::max(cpus / 2, MIN_RUNTIME_THREADS),
options.compact_rt_max_blocking_threads
);
assert_eq!(usize::max(cpus.saturating_sub(1), 1), options.query_rt_size);
assert_eq!(
usize::max(cpus.saturating_sub(1), MIN_RUNTIME_THREADS),
options.query_rt_size
);
assert_eq!(cpus, options.ingest_rt_size);
}
#[test]
fn test_runtime_options_min_threads() {
for cpus in [0, 1, 2] {
let options = RuntimeOptions::with_num_cpus(cpus);
assert!(
options.global_rt_size >= MIN_RUNTIME_THREADS,
"global_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
options.global_rt_size
);
assert!(
options.compact_rt_size >= MIN_RUNTIME_THREADS,
"compact_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
options.compact_rt_size
);
assert!(
options.compact_rt_max_blocking_threads >= MIN_RUNTIME_THREADS,
"compact_rt_max_blocking_threads {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
options.compact_rt_max_blocking_threads
);
assert!(
options.query_rt_size >= MIN_RUNTIME_THREADS,
"query_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
options.query_rt_size
);
assert!(
options.ingest_rt_size >= MIN_RUNTIME_THREADS,
"ingest_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
options.ingest_rt_size
);
}
}
#[test]
fn test_datanode_runtimes_fallback_to_global_runtime() {
let runtimes = GlobalRuntimes::new(
+3
View File
@@ -27,6 +27,9 @@ jsonb.workspace = true
num = "0.4"
num-traits = "0.2"
ordered-float.workspace = true
parquet-variant.workspace = true
parquet-variant-compute.workspace = true
parquet-variant-json.workspace = true
paste.workspace = true
regex.workspace = true
serde.workspace = true
+16
View File
@@ -77,6 +77,13 @@ pub enum Error {
location: Location,
},
#[snafu(display("Unimplemented: {feat}"))]
Unimplemented {
feat: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to parse version in schema meta, value: {}", value))]
ParseSchemaVersion {
value: String,
@@ -196,6 +203,13 @@ pub enum Error {
location: Location,
},
#[snafu(display("Invalid JSON2 layout: {reason}"))]
InvalidJson2Layout {
reason: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Invalid Vector: {}", msg))]
InvalidVector {
msg: String,
@@ -305,6 +319,7 @@ impl ErrorExt for Error {
use Error::*;
match self {
UnsupportedOperation { .. }
| Unimplemented { .. }
| UnsupportedArrowType { .. }
| UnsupportedJsonType { .. }
| UnsupportedDefaultExpr { .. } => StatusCode::Unsupported,
@@ -318,6 +333,7 @@ impl ErrorExt for Error {
| InvalidTimestampPrecision { .. }
| InvalidPrecisionOrScale { .. }
| InvalidJson { .. }
| InvalidJson2Layout { .. }
| InvalidJsonb { .. }
| InvalidVector { .. }
| InvalidFulltextOption { .. }
+198 -5
View File
@@ -18,30 +18,148 @@ use std::sync::Arc;
use arrow_schema::extension::{
EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY, ExtensionType,
};
use arrow_schema::{ArrowError, DataType, Field};
use arrow_schema::{ArrowError, DataType, Field, FieldRef};
use parquet_variant_compute::VariantType;
use serde::{Deserialize, Serialize};
use snafu::ResultExt;
use snafu::{ResultExt, ensure};
use crate::error::InvalidJson2LayoutSnafu;
pub use crate::json::JSON2_REMAINDER_FIELD_NAME;
use crate::json::JsonSettings;
const LEGACY_JSON_STRUCTURE_SETTINGS_KEY: &str = "json_structure_settings";
const JSON2_LAYOUT_V1: u8 = 1;
const JSON2_LAYOUT_V2: u8 = 2;
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
/// Parsed physical layout of a JSON2 Arrow root field.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Json2PhysicalLayout {
version: u8,
}
impl Json2PhysicalLayout {
/// Parses the JSON2 layout version from a root field.
///
/// The version is read from the extension metadata; missing metadata
/// defaults to V1.
///
/// Supported versions:
/// - V1
/// - V2
///
/// Errors when the field is not a JSON2 extension or the version is not
/// one of the supported versions.
pub fn try_from_root(field: &Field) -> crate::error::Result<Self> {
ensure!(
is_json2_extension_type(field),
InvalidJson2LayoutSnafu {
reason: format!("field '{}' is not a JSON2 extension", field.name()),
}
);
let version = match field.extension_type_name() {
Some(Json2ExtensionType::NAME) => field
.metadata()
.get(EXTENSION_TYPE_METADATA_KEY)
.map(|x| parse_version(x))
.transpose()?
.flatten()
.unwrap_or(JSON2_LAYOUT_V1),
_ => JSON2_LAYOUT_V1,
};
ensure!(
matches!(version, JSON2_LAYOUT_V1 | JSON2_LAYOUT_V2),
InvalidJson2LayoutSnafu {
reason: format!("unsupported JSON2 layout version: {version}"),
}
);
Ok(Self { version })
}
/// Returns whether this is the JSON2 physical layout version 2.
pub fn is_version_2(&self) -> bool {
self.version == JSON2_LAYOUT_V2
}
}
fn parse_version(metadata: &str) -> crate::error::Result<Option<u8>> {
serde_json::from_str::<JsonMetadata>(metadata)
.map(|x| x.layout_version)
.map_err(|e| {
InvalidJson2LayoutSnafu {
reason: format!(r#"invalid extension metadata: "{metadata}", error: {e}"#),
}
.build()
})
}
/// Returns the remainder field of a JSON2 v2 root.
pub(crate) fn json2_remainder_field(field: &Field) -> crate::error::Result<Option<&FieldRef>> {
let layout = Json2PhysicalLayout::try_from_root(field)?;
if !layout.is_version_2() {
return Ok(None);
}
let remainder = if let DataType::Struct(fields) = field.data_type() {
fields
.iter()
.find(|x| x.name() == JSON2_REMAINDER_FIELD_NAME)
} else {
None
};
if let Some(remainder) = remainder {
let _ = remainder.try_extension_type::<VariantType>().map_err(|e| {
InvalidJson2LayoutSnafu {
reason: e.to_string(),
}
.build()
})?;
}
Ok(remainder)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct JsonMetadata {
/// JSON2 settings stored in Arrow extension metadata.
json_settings: JsonSettings,
/// Physical JSON2 layout used by this Arrow field.
#[serde(default, skip_serializing_if = "Option::is_none")]
layout_version: Option<u8>,
}
impl JsonMetadata {
/// Creates JSON2 extension metadata.
/// Creates metadata for the legacy JSON2 layout.
pub fn new(json_settings: JsonSettings) -> Self {
Self { json_settings }
Self {
json_settings,
layout_version: None,
}
}
/// Creates metadata for the new JSON2 physical v2 layout.
pub fn new_v2(json_settings: JsonSettings) -> Self {
Self {
json_settings,
layout_version: Some(JSON2_LAYOUT_V2),
}
}
/// Returns the JSON2 settings.
pub fn json_settings(&self) -> &JsonSettings {
&self.json_settings
}
/// Returns whether this metadata describes JSON2 layout version 2.
pub fn is_version_2(&self) -> bool {
self.layout_version == Some(JSON2_LAYOUT_V2)
}
}
impl Default for JsonMetadata {
fn default() -> Self {
Self::new(JsonSettings::default())
}
}
/// Arrow extension type for legacy JSONB columns.
@@ -226,6 +344,7 @@ mod tests {
use arrow_schema::{Field, Fields};
use super::*;
use crate::vectors::json::variant::variant_field;
#[test]
fn test_json2_extension_type_detection() {
@@ -281,4 +400,78 @@ mod tests {
assert!(JsonExtensionType::try_new(&DataType::Null, ()).is_ok());
assert!(JsonExtensionType::try_new(&DataType::Struct(Fields::empty()), ()).is_err());
}
#[test]
fn test_json_metadata_layout_version_compatibility() -> serde_json::Result<()> {
let legacy: JsonMetadata = serde_json::from_str(r#"{"json_settings":{}}"#)?;
assert!(!legacy.is_version_2());
let metadata = JsonMetadata::new_v2(JsonSettings::default());
assert!(metadata.is_version_2());
let serialized = serde_json::to_string(&metadata)?;
let deserialized: JsonMetadata = serde_json::from_str(&serialized)?;
assert!(deserialized.is_version_2());
assert_eq!(deserialized, metadata);
Ok(())
}
#[test]
fn test_parse_json2_physical_layout() -> crate::error::Result<()> {
let legacy = Field::new("data", DataType::Struct(Fields::empty()), true)
.with_extension_type(Json2ExtensionType::default());
assert!(!Json2PhysicalLayout::try_from_root(&legacy)?.is_version_2());
let v2 = Field::new(
"data",
DataType::Struct(
vec![
Arc::new(variant_field(JSON2_REMAINDER_FIELD_NAME, true)),
Arc::new(Field::new("count", DataType::Int64, true)),
]
.into(),
),
true,
)
.with_extension_type(Json2ExtensionType::new(Arc::new(JsonMetadata::new_v2(
JsonSettings::default(),
))));
assert!(Json2PhysicalLayout::try_from_root(&v2)?.is_version_2());
let remainder = json2_remainder_field(&v2)?.unwrap();
assert_eq!(JSON2_REMAINDER_FIELD_NAME, remainder.name());
Ok(())
}
#[test]
fn test_reject_invalid_json2_physical_layout() {
let field = Field::new("data", DataType::Struct(Fields::empty()), true);
assert!(Json2PhysicalLayout::try_from_root(&field).is_err());
let metadata =
Json2ExtensionType::new(Arc::new(JsonMetadata::new_v2(JsonSettings::default())));
let missing = Field::new("data", DataType::Struct(Fields::empty()), true)
.with_extension_type(metadata.clone());
assert!(json2_remainder_field(&missing).is_ok_and(|x| x.is_none()));
let invalid = Field::new(
"data",
DataType::Struct(
vec![Arc::new(Field::new(
JSON2_REMAINDER_FIELD_NAME,
DataType::Binary,
true,
))]
.into(),
),
true,
)
.with_extension_type(metadata);
assert!(json2_remainder_field(&invalid).is_err());
let future = Field::new("data", DataType::Struct(Fields::empty()), true)
.with_extension_type(Json2ExtensionType::new(Arc::new(JsonMetadata {
json_settings: JsonSettings::default(),
layout_version: Some(JSON2_LAYOUT_V2 + 1),
})));
assert!(Json2PhysicalLayout::try_from_root(&future).is_err());
}
}
+269 -42
View File
@@ -29,21 +29,47 @@ use serde_json::{Map, Value as Json};
use snafu::ResultExt;
use crate::data_type::ConcreteDataType;
use crate::error::{self, Result, UnsupportedJsonTypeSnafu};
use crate::error::{self, InvalidJson2LayoutSnafu, Result, UnsupportedJsonTypeSnafu};
use crate::json::value::{JsonValue, JsonVariant, encode_serde_json_as_jsonb};
use crate::schema::ColumnDefaultConstraint;
use crate::types::json_type::JsonNativeType;
use crate::types::json_type::{JsonNativeType, JsonObjectType};
use crate::value::{ListValue, StructValue, Value};
/// Maximum number of JSON container levels represented as nested Arrow types.
pub const JSON2_MAX_STRUCTURED_DEPTH: usize = 50;
/// Reserved physical field containing unexpanded JSON2 paths.
pub const JSON2_REMAINDER_FIELD_NAME: &str = "!__remainder__!";
/// JSON2 settings stored in column schema metadata and represented through
/// Arrow extension metadata.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct JsonSettings {
#[serde(default)]
pub type_hints: Vec<JsonTypeHint>,
type_hints: Vec<JsonTypeHint>,
/// Maximum number of unhinted JSON paths expanded into Arrow fields.
///
/// `None` preserves the legacy unlimited layout.
#[serde(default, skip_serializing_if = "Option::is_none")]
max_auto_expanded_paths: Option<u32>,
}
#[derive(Deserialize)]
struct JsonSettingsSerde {
#[serde(default)]
type_hints: Vec<JsonTypeHint>,
#[serde(default)]
max_auto_expanded_paths: Option<u32>,
}
impl<'de> Deserialize<'de> for JsonSettings {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let settings = JsonSettingsSerde::deserialize(deserializer)?;
Self::try_new(settings.type_hints, settings.max_auto_expanded_paths)
.map_err(serde::de::Error::custom)
}
}
/// Declares selected JSON2 subpaths as typed fields.
@@ -79,8 +105,28 @@ pub struct JsonContext<'a> {
}
impl JsonSettings {
pub fn new(type_hints: Vec<JsonTypeHint>) -> Self {
Self { type_hints }
/// Creates and validates JSON2 settings.
pub fn try_new(
type_hints: Vec<JsonTypeHint>,
max_auto_expanded_paths: Option<u32>,
) -> Result<Self> {
validate_type_hints(&type_hints)?;
Ok(Self {
type_hints,
max_auto_expanded_paths,
})
}
pub fn type_hints(&self) -> &[JsonTypeHint] {
&self.type_hints
}
pub fn max_auto_expanded_paths(&self) -> Option<u32> {
self.max_auto_expanded_paths
}
pub fn into_parts(self) -> (Vec<JsonTypeHint>, Option<u32>) {
(self.type_hints, self.max_auto_expanded_paths)
}
/// Decode an encoded StructValue back into a serde_json::Value.
@@ -94,6 +140,16 @@ impl JsonSettings {
/// Encode a serde_json::Value into a Value::Json using current settings.
pub fn encode(&self, json: Json) -> Result<Value> {
if let Json::Object(object) = &json
&& object.contains_key(JSON2_REMAINDER_FIELD_NAME)
{
return error::InvalidJsonSnafu {
value: format!(
"root object cannot contain reserved field '{JSON2_REMAINDER_FIELD_NAME}'"
),
}
.fail();
}
let mut context = JsonContext {
path: Vec::new(),
settings: self,
@@ -102,6 +158,87 @@ impl JsonSettings {
}
}
fn validate_type_hints(type_hints: &[JsonTypeHint]) -> Result<()> {
let mut object = JsonObjectType::new();
for hint in type_hints {
if hint.path.len() > JSON2_MAX_STRUCTURED_DEPTH {
return InvalidJson2LayoutSnafu {
reason: format!(
"JSON2 type hint path cannot exceed {JSON2_MAX_STRUCTURED_DEPTH} segments"
),
}
.fail();
}
if hint
.path
.first()
.is_some_and(|x| x == JSON2_REMAINDER_FIELD_NAME)
{
return InvalidJson2LayoutSnafu {
reason: format!(
"JSON2 type hint path cannot be rooted at reserved field '{JSON2_REMAINDER_FIELD_NAME}'"
),
}
.fail();
}
let data_type = match &hint.data_type {
ConcreteDataType::Boolean(_)
| ConcreteDataType::UInt8(_)
| ConcreteDataType::UInt16(_)
| ConcreteDataType::UInt32(_)
| ConcreteDataType::UInt64(_)
| ConcreteDataType::Int8(_)
| ConcreteDataType::Int16(_)
| ConcreteDataType::Int32(_)
| ConcreteDataType::Int64(_)
| ConcreteDataType::Float32(_)
| ConcreteDataType::Float64(_)
| ConcreteDataType::String(_) => (&hint.data_type).into(),
data_type => {
return InvalidJson2LayoutSnafu {
reason: format!("unsupported JSON2 type hint data type: {data_type}"),
}
.fail();
}
};
validate_type_hint(&mut object, &hint.path, data_type)?;
}
Ok(())
}
fn validate_type_hint(
object: &mut JsonObjectType,
path: &[String],
data_type: JsonNativeType,
) -> Result<()> {
let Some((name, path)) = path.split_first() else {
return InvalidJson2LayoutSnafu {
reason: "JSON2 type hint path must not be empty".to_string(),
}
.fail();
};
if path.is_empty() {
if object.insert(name.clone(), data_type).is_some() {
return InvalidJson2LayoutSnafu {
reason: format!("duplicate JSON2 type hint path '{name}'"),
}
.fail();
}
return Ok(());
}
let child = object
.entry(name.clone())
.or_insert_with(|| JsonNativeType::Object(JsonObjectType::new()));
let JsonNativeType::Object(child) = child else {
return InvalidJson2LayoutSnafu {
reason: format!("conflicting JSON2 type hint path at '{name}'"),
}
.fail();
};
validate_type_hint(child, path, data_type)
}
impl<'a> JsonContext<'a> {
fn type_hint(&self) -> Option<&'a JsonTypeHint> {
self.settings
@@ -483,7 +620,8 @@ mod tests {
}
#[test]
fn test_json_settings_forward_compatibility() {
fn test_json_settings_forward_compatibility()
-> std::result::Result<(), Box<dyn std::error::Error>> {
let json_str = r#"{
"type_hints": [
{
@@ -512,11 +650,10 @@ mod tests {
]
}"#;
let deserialized = serde_json::from_str::<JsonSettings>(json_str).unwrap();
let deserialized = serde_json::from_str::<JsonSettings>(json_str)?;
assert_eq!(
deserialized,
JsonSettings::new(vec![
let expected = JsonSettings::try_new(
vec![
JsonTypeHint {
path: vec!["user".to_string(), "age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
@@ -531,13 +668,16 @@ mod tests {
default_constraint: None,
inverted_index: false,
},
])
);
],
None,
)?;
assert_eq!(deserialized, expected);
Ok(())
}
#[test]
fn test_json_settings_ser_de() {
let settings = JsonSettings::new(vec![
fn test_json_settings_ser_de() -> std::result::Result<(), Box<dyn std::error::Error>> {
let type_hints = vec![
JsonTypeHint {
path: vec!["user".to_string(), "age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
@@ -552,12 +692,76 @@ mod tests {
default_constraint: None,
inverted_index: false,
},
]);
];
let serialized = serde_json::to_string(&settings).unwrap();
let deserialized = serde_json::from_str::<JsonSettings>(&serialized).unwrap();
for settings in [
JsonSettings {
type_hints: type_hints.clone(),
max_auto_expanded_paths: None,
},
JsonSettings {
type_hints,
max_auto_expanded_paths: Some(1),
},
] {
let serialized = serde_json::to_string(&settings)?;
let deserialized = serde_json::from_str::<JsonSettings>(&serialized)?;
assert_eq!(settings, deserialized);
}
Ok(())
}
assert_eq!(settings, deserialized);
#[test]
fn test_json_settings_reject_invalid_type_hint_layout() {
for type_hints in [
json!([{"path": [], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}]),
json!([
{"path": ["a"], "type": {"Int64": {}}, "nullable": true, "inverted_index": false},
{"path": ["a", "b"], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}
]),
json!([{"path": [JSON2_REMAINDER_FIELD_NAME], "type": {"Int64": {}}, "nullable": true, "inverted_index": false}]),
] {
let settings = json!({"type_hints": type_hints});
assert!(serde_json::from_value::<JsonSettings>(settings).is_err());
}
let hint = |path, data_type| JsonTypeHint {
path,
data_type,
nullable: true,
default_constraint: None,
inverted_index: false,
};
assert!(
JsonSettings::try_new(
vec![hint(
vec!["nested".to_string(), JSON2_REMAINDER_FIELD_NAME.to_string()],
ConcreteDataType::string_datatype(),
)],
None,
)
.is_ok()
);
assert!(
JsonSettings::try_new(
vec![hint(
vec!["nested".to_string(); JSON2_MAX_STRUCTURED_DEPTH + 1],
ConcreteDataType::string_datatype(),
)],
None,
)
.is_err()
);
assert!(
JsonSettings::try_new(
vec![hint(
vec!["date".to_string()],
ConcreteDataType::date_datatype(),
)],
None,
)
.is_err()
);
}
#[test]
@@ -583,6 +787,21 @@ mod tests {
}
}
#[test]
fn test_encode_rejects_reserved_remainder_field() -> Result<()> {
let settings = JsonSettings::default();
let err = settings
.encode(json!({"!__remainder__!": "user-value"}))
.unwrap_err();
assert!(
err.to_string()
.contains("root object cannot contain reserved field")
);
settings.encode(json!({"nested": {"!__remainder__!": "user-value"}}))?;
Ok(())
}
#[test]
fn test_encode_json_object() {
let json = json!({
@@ -702,14 +921,15 @@ mod tests {
}
#[test]
fn test_encode_json_respects_type_hint() {
let settings = JsonSettings::new(vec![JsonTypeHint {
fn test_encode_json_respects_type_hint() -> Result<()> {
let type_hints = vec![JsonTypeHint {
path: vec!["age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
nullable: false,
default_constraint: None,
inverted_index: false,
}]);
}];
let settings = JsonSettings::try_new(type_hints, None)?;
let result = settings
.encode(json!({
@@ -731,17 +951,19 @@ mod tests {
}))
.unwrap_err();
assert!(err.to_string().contains("does not match JSON2 type hint"));
Ok(())
}
#[test]
fn test_encode_json_respects_unsigned_type_hint() {
let settings = JsonSettings::new(vec![JsonTypeHint {
fn test_encode_json_respects_unsigned_type_hint() -> Result<()> {
let type_hints = vec![JsonTypeHint {
path: vec!["count".to_string()],
data_type: ConcreteDataType::uint64_datatype(),
nullable: false,
default_constraint: None,
inverted_index: false,
}]);
}];
let settings = JsonSettings::try_new(type_hints, None)?;
let result = settings
.encode(json!({
@@ -765,17 +987,19 @@ mod tests {
}))
.unwrap_err();
assert!(err.to_string().contains("does not match JSON2 type hint"));
Ok(())
}
#[test]
fn test_encode_json_fills_missing_type_hint_with_default() {
let settings = JsonSettings::new(vec![JsonTypeHint {
fn test_encode_json_fills_missing_type_hint_with_default() -> Result<()> {
let type_hints = vec![JsonTypeHint {
path: vec!["user".to_string(), "age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
nullable: false,
default_constraint: Some(ColumnDefaultConstraint::Value(Value::Int64(7))),
inverted_index: false,
}]);
}];
let settings = JsonSettings::try_new(type_hints, None)?;
let result = settings
.encode(json!({}))
@@ -790,17 +1014,19 @@ mod tests {
panic!("Expected user Struct value");
};
assert_eq!(struct_field_value(user, "age"), &Value::Int64(7));
Ok(())
}
#[test]
fn test_encode_json_fills_missing_nullable_type_hint_with_null() {
let settings = JsonSettings::new(vec![JsonTypeHint {
fn test_encode_json_fills_missing_nullable_type_hint_with_null() -> Result<()> {
let type_hints = vec![JsonTypeHint {
path: vec!["user".to_string(), "name".to_string()],
data_type: ConcreteDataType::string_datatype(),
nullable: true,
default_constraint: None,
inverted_index: false,
}]);
}];
let settings = JsonSettings::try_new(type_hints, None)?;
let result = settings
.encode(json!({ "user": {} }))
@@ -815,28 +1041,31 @@ mod tests {
panic!("Expected user Struct value");
};
assert_eq!(struct_field_value(user, "name"), &Value::Null);
Ok(())
}
#[test]
fn test_encode_json_rejects_missing_non_null_type_hint() {
let settings = JsonSettings::new(vec![JsonTypeHint {
fn test_encode_json_rejects_missing_non_null_type_hint() -> Result<()> {
let type_hints = vec![JsonTypeHint {
path: vec!["user".to_string(), "age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
nullable: false,
default_constraint: None,
inverted_index: false,
}]);
}];
let settings = JsonSettings::try_new(type_hints, None)?;
let err = settings.encode(json!({})).unwrap_err();
assert!(
err.to_string()
.contains("missing non-null JSON2 type hint path user.age")
);
Ok(())
}
#[test]
fn test_encode_json_merges_missing_type_hint_prefix() {
let settings = JsonSettings::new(vec![
fn test_encode_json_merges_missing_type_hint_prefix() -> Result<()> {
let type_hints = vec![
JsonTypeHint {
path: vec!["user".to_string(), "age".to_string()],
data_type: ConcreteDataType::int64_datatype(),
@@ -853,13 +1082,10 @@ mod tests {
))),
inverted_index: false,
},
]);
];
let settings = JsonSettings::try_new(type_hints, None)?;
let result = settings
.encode(json!({}))
.unwrap()
.into_json_inner()
.unwrap();
let result = settings.encode(json!({}))?.into_json_inner().unwrap();
let Value::Struct(root) = result else {
panic!("Expected Struct value");
@@ -872,6 +1098,7 @@ mod tests {
struct_field_value(user, "name"),
&Value::String("unknown".into())
);
Ok(())
}
#[test]
+1
View File
@@ -14,3 +14,4 @@
pub mod array;
pub(crate) mod builder;
pub mod variant;
+235 -13
View File
@@ -17,9 +17,12 @@ use std::sync::Arc;
use arrow::compute::{can_cast_types, cast};
use arrow_array::cast::AsArray;
use arrow_array::types::{Float64Type, Int64Type, UInt64Type};
use arrow_array::types::{
Float32Type, Float64Type, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type,
UInt32Type, UInt64Type,
};
use arrow_array::{Array, ArrayRef, GenericListArray, ListArray, StructArray, new_null_array};
use arrow_schema::{DataType, FieldRef};
use arrow_schema::{DataType, Field, FieldRef};
use common_telemetry::trace;
use serde_json::Value;
use snafu::{OptionExt, ResultExt};
@@ -29,9 +32,11 @@ use crate::data_type::ConcreteDataType;
use crate::error::{
AlignJsonArraySnafu, ArrowComputeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, Result,
};
use crate::extension::json::{JSON2_REMAINDER_FIELD_NAME, json2_remainder_field};
use crate::json::value::{decode_json_variant, encode_serde_json_as_jsonb};
use crate::prelude::{DataType as _, Value as GreptimeValue};
use crate::value::{ListValue, StructValue};
use crate::vectors::json::variant::variant_to_json_values;
pub struct JsonArray<'a> {
inner: &'a ArrayRef,
@@ -48,8 +53,15 @@ impl JsonArray<'_> {
let value = match array.data_type() {
DataType::Null => Value::Null,
DataType::Boolean => Value::Bool(array.as_boolean().value(i)),
DataType::Int8 => Value::from(array.as_primitive::<Int8Type>().value(i)),
DataType::Int16 => Value::from(array.as_primitive::<Int16Type>().value(i)),
DataType::Int32 => Value::from(array.as_primitive::<Int32Type>().value(i)),
DataType::Int64 => Value::from(array.as_primitive::<Int64Type>().value(i)),
DataType::UInt8 => Value::from(array.as_primitive::<UInt8Type>().value(i)),
DataType::UInt16 => Value::from(array.as_primitive::<UInt16Type>().value(i)),
DataType::UInt32 => Value::from(array.as_primitive::<UInt32Type>().value(i)),
DataType::UInt64 => Value::from(array.as_primitive::<UInt64Type>().value(i)),
DataType::Float32 => Value::from(array.as_primitive::<Float32Type>().value(i)),
DataType::Float64 => Value::from(array.as_primitive::<Float64Type>().value(i)),
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => {
Value::String(string_array_value(array, i).to_string())
@@ -92,6 +104,60 @@ impl JsonArray<'_> {
Ok(value)
}
/// Projects a physical JSON2 array to a logical query type.
///
/// TODO(LFC) Supersede `project_to_v2` to `project_to`.
pub fn project_to_v2(&self, field: &Field, target: &DataType) -> Result<ArrayRef> {
if json2_remainder_field(field)?.is_some() {
project_json_values(self.json2_values()?, target)
} else {
self.project_to(target)
}
}
fn json2_values(&self) -> Result<Vec<Value>> {
let structs = self.inner.as_struct_opt().context(AlignJsonArraySnafu {
reason: "JSON2 layout v2 root array must be a struct",
})?;
let remainder = structs.column_by_name(JSON2_REMAINDER_FIELD_NAME);
let mut remainders = if let Some(remainder) = remainder {
variant_to_json_values(remainder)?
} else {
vec![None; structs.len()]
};
let mut values = Vec::with_capacity(structs.len());
let mut path = Vec::new();
for (i, remainder) in remainders.iter_mut().enumerate() {
if structs.is_null(i) {
values.push(Value::Null);
continue;
}
let mut object = match remainder.take() {
None => serde_json::Map::new(),
Some(Value::Object(object)) => object,
Some(value) => {
return InvalidJsonSnafu {
value: format!("JSON2 layout v2 remainder must be an object, got {value}"),
}
.fail();
}
};
for (child, column) in structs.fields().iter().zip(structs.columns()) {
if child.name() == JSON2_REMAINDER_FIELD_NAME {
continue;
}
let value = JsonArray::from(column).try_get_value(i)?;
merge_explicit_value(&mut object, child.name().clone(), value, &mut path)?;
}
values.push(Value::Object(object));
}
Ok(values)
}
/// Normalizes a JSON2 array to the wider `expect` data type without losing
/// information.
///
@@ -326,6 +392,34 @@ impl JsonArray<'_> {
}
}
fn merge_explicit_value(
remainder: &mut serde_json::Map<String, Value>,
key: String,
explicit: Value,
path: &mut Vec<String>,
) -> Result<()> {
let Some(existing) = remainder.get_mut(&key) else {
remainder.insert(key, explicit);
return Ok(());
};
path.push(key);
let (Value::Object(remainder), Value::Object(explicit)) = (existing, explicit) else {
return InvalidJsonSnafu {
value: format!(
"cannot merge '{}' in explicit fields and remainder: not both objects",
path.join("."),
),
}
.fail();
};
for (key, value) in explicit {
merge_explicit_value(remainder, key, value, path)?;
}
path.pop();
Ok(())
}
/// Returns whether Arrow can cast between the types without JSON-aware projection.
/// Binary and nested types require JSONB decoding or recursive projection.
fn can_fast_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
@@ -453,13 +547,16 @@ impl<'a> From<&'a ArrayRef> for JsonArray<'a> {
mod test {
use arrow_array::types::Int64Type;
use arrow_array::{
BinaryArray, BooleanArray, Float64Array, Int32Array, Int64Array, ListArray, StringArray,
UInt64Array,
BinaryArray, BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array,
Int64Array, ListArray, StringArray, UInt8Array, UInt16Array, UInt32Array, UInt64Array,
};
use arrow_schema::{Field, Fields};
use serde_json::json;
use super::*;
use crate::extension::json::{Json2ExtensionType, JsonMetadata};
use crate::json::JsonSettings;
use crate::vectors::json::variant::{json_values_to_variant, variant_field};
#[test]
fn test_try_get_value() -> Result<()> {
@@ -474,6 +571,20 @@ mod test {
assert_eq!(JsonArray::from(&ints).try_get_value(0)?, json!(-7));
assert_eq!(JsonArray::from(&ints).try_get_value(1)?, Value::Null);
macro_rules! assert_number {
($array:expr, $expected:expr) => {{
let array: ArrayRef = Arc::new($array);
assert_eq!(JsonArray::from(&array).try_get_value(0)?, json!($expected));
}};
}
assert_number!(Int8Array::from(vec![-8]), -8);
assert_number!(Int16Array::from(vec![-16]), -16);
assert_number!(Int32Array::from(vec![-32]), -32);
assert_number!(UInt8Array::from(vec![8]), 8);
assert_number!(UInt16Array::from(vec![16]), 16);
assert_number!(UInt32Array::from(vec![32]), 32);
assert_number!(Float32Array::from(vec![1.25]), 1.25);
let floats: ArrayRef = Arc::new(Float64Array::from(vec![Some(1.5)]));
assert_eq!(JsonArray::from(&floats).try_get_value(0)?, json!(1.5));
@@ -529,15 +640,6 @@ mod test {
json!({"flag": null, "items": [2]})
);
let unsupported: ArrayRef = Arc::new(Int32Array::from(vec![1]));
assert_eq!(
JsonArray::from(&unsupported)
.try_get_value(0)
.unwrap_err()
.to_string(),
"Invalid JSON: unknown JSON type Int32"
);
Ok(())
}
@@ -915,4 +1017,124 @@ mod test {
Ok(())
}
#[test]
fn test_reconstruct_json2_v2_value() -> Result<()> {
let remainders = json_values_to_variant(&[
Some(json!({"cold": 1, "nested": {"right": true}})),
Some(json!({"!__remainder__!": "user value"})),
])?;
let remainder = Arc::new(variant_field(JSON2_REMAINDER_FIELD_NAME, true));
let nested = Arc::new(Field::new_struct(
"nested",
[Arc::new(Field::new("left", DataType::Utf8, true))],
true,
));
let nested_values: ArrayRef = Arc::new(StructArray::from(vec![(
Arc::new(Field::new("left", DataType::Utf8, true)),
Arc::new(StringArray::from(vec![Some("value"), None])) as ArrayRef,
)]));
let fields = Fields::from(vec![
remainder,
Arc::new(Field::new("count", DataType::Int64, true)),
nested,
]);
let array: ArrayRef = Arc::new(StructArray::new(
fields.clone(),
vec![
remainders,
Arc::new(Int64Array::from(vec![Some(42), None])),
nested_values,
],
None,
));
let field = Field::new("data", DataType::Struct(fields), true).with_extension_type(
Json2ExtensionType::new(Arc::new(JsonMetadata::new_v2(JsonSettings::default()))),
);
assert_eq!(
json!({
"cold": 1,
"count": 42,
"nested": {"left": "value", "right": true}
}),
JsonArray::from(&array).json2_values()?[0]
);
assert_eq!(
json!({
"!__remainder__!": "user value",
"count": null,
"nested": {"left": null}
}),
JsonArray::from(&array).json2_values()?[1]
);
let target = DataType::Struct(
vec![
Arc::new(Field::new("cold", DataType::UInt64, true)),
Arc::new(Field::new("count", DataType::Int64, true)),
]
.into(),
);
let projected = JsonArray::from(&array).project_to_v2(&field, &target)?;
assert_eq!(
json!({"cold": 1, "count": 42}),
JsonArray::from(&projected).try_get_value(0)?
);
assert_eq!(
json!({"cold": null, "count": null}),
JsonArray::from(&projected).try_get_value(1)?
);
Ok(())
}
#[test]
fn test_project_partial_json2_v2_without_remainder() -> Result<()> {
let fields = Fields::from(vec![Arc::new(Field::new("hot", DataType::Int64, true))]);
let array: ArrayRef = Arc::new(StructArray::new(
fields.clone(),
vec![Arc::new(Int64Array::from(vec![1, 2]))],
None,
));
let field = Field::new("data", DataType::Struct(fields), true).with_extension_type(
Json2ExtensionType::new(Arc::new(JsonMetadata::new_v2(JsonSettings::default()))),
);
let projected = JsonArray::from(&array).project_to_v2(&field, field.data_type())?;
assert!(Arc::ptr_eq(&array, &projected));
Ok(())
}
#[test]
fn test_reject_conflict_json2_v2_path() -> Result<()> {
let remainders = json_values_to_variant(&[Some(json!({"count": 1}))])?;
let fields = Fields::from(vec![
Arc::new(variant_field(JSON2_REMAINDER_FIELD_NAME, true)),
Arc::new(Field::new("count", DataType::Int64, true)),
]);
let array: ArrayRef = Arc::new(StructArray::new(
fields,
vec![remainders, Arc::new(Int64Array::from(vec![2]))],
None,
));
let error = JsonArray::from(&array).json2_values().unwrap_err();
assert!(
error.to_string().contains(
"cannot merge 'count' in explicit fields and remainder: not both objects"
)
);
let Value::Object(mut remainder) = json!({"nested": {"count": 1}}) else {
unreachable!();
};
let error = merge_explicit_value(
&mut remainder,
"nested".to_string(),
json!({"count": 2}),
&mut Vec::new(),
)
.unwrap_err();
assert!(error.to_string().contains("cannot merge 'nested.count'"));
Ok(())
}
}
File diff suppressed because it is too large Load Diff
+278
View File
@@ -0,0 +1,278 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::sync::Arc;
use arrow_array::ArrayRef;
use arrow_schema::{ArrowError, DataType, Field};
use parquet_variant::{ObjectFieldBuilder, Variant, VariantBuilderExt, VariantDecimal16};
#[cfg(test)]
use parquet_variant_compute::VariantArrayBuilder;
use parquet_variant_compute::{VariantArray, VariantType};
use parquet_variant_json::VariantToJson;
use snafu::ResultExt;
use crate::error::{ArrowComputeSnafu, Result};
use crate::json::value::{JsonNumber, JsonVariant, JsonVariantRef, decode_json_variant};
/// Returns the canonical Arrow field for an unshredded Parquet Variant array.
pub fn variant_field(name: impl Into<String>, nullable: bool) -> Field {
Field::new_struct(
name,
[
Arc::new(Field::new("metadata", DataType::BinaryView, false)),
Arc::new(Field::new("value", DataType::BinaryView, false)),
],
nullable,
)
.with_extension_type(VariantType)
}
#[cfg(test)]
pub(crate) fn json_values_to_variant(values: &[Option<serde_json::Value>]) -> Result<ArrayRef> {
let mut builder = VariantArrayBuilder::new(values.len());
for value in values {
match value {
Some(value) => append_json_value(&mut builder, value).context(ArrowComputeSnafu)?,
None => builder.append_null(),
}
}
Ok(ArrayRef::from(builder.build()))
}
/// Encodes JSON variants as an unshredded Parquet Variant array.
#[cfg(test)]
fn json_variants_to_variant(values: &[Option<JsonVariant>]) -> Result<ArrayRef> {
let mut builder = VariantArrayBuilder::new(values.len());
for value in values {
match value {
Some(value) => append_json_variant(&mut builder, value).context(ArrowComputeSnafu)?,
None => builder.append_null(),
}
}
Ok(ArrayRef::from(builder.build()))
}
pub(super) fn append_json_variant(
builder: &mut impl VariantBuilderExt,
value: &JsonVariant,
) -> std::result::Result<(), ArrowError> {
match value {
JsonVariant::Null => builder.append_value(Variant::Null),
JsonVariant::Bool(value) => builder.append_value(*value),
JsonVariant::Number(JsonNumber::PosInt(value)) => {
if let Ok(value) = i64::try_from(*value) {
builder.append_value(value);
} else {
append_large_u64(builder, *value)?;
}
}
JsonVariant::Number(JsonNumber::NegInt(value)) => builder.append_value(*value),
JsonVariant::Number(JsonNumber::Float(value)) => {
if value.0.is_finite() {
builder.append_value(value.0)
} else {
builder.append_value("NaN")
}
}
JsonVariant::String(value) => builder.append_value(value.as_str()),
JsonVariant::Array(values) => {
let mut list = builder.try_new_list()?;
for value in values {
append_json_variant(&mut list, value)?;
}
list.finish();
}
JsonVariant::Object(values) => {
let mut object = builder.try_new_object()?;
for (name, value) in values {
append_json_variant(&mut ObjectFieldBuilder::new(name, &mut object), value)?;
}
object.finish();
}
JsonVariant::Variant(value) => {
let value = decode_json_variant(value)
.map_err(|e| ArrowError::JsonError(format!("Failed to decode JSONB: {e}")))?;
append_json_value(builder, &value)?;
}
}
Ok(())
}
pub(super) fn append_json_variant_ref(
builder: &mut impl VariantBuilderExt,
value: &JsonVariantRef<'_>,
) -> std::result::Result<(), ArrowError> {
match value {
JsonVariantRef::Null => builder.append_value(Variant::Null),
JsonVariantRef::Bool(value) => builder.append_value(*value),
JsonVariantRef::Number(JsonNumber::PosInt(value)) => {
if let Ok(value) = i64::try_from(*value) {
builder.append_value(value);
} else {
append_large_u64(builder, *value)?;
}
}
JsonVariantRef::Number(JsonNumber::NegInt(value)) => builder.append_value(*value),
JsonVariantRef::Number(JsonNumber::Float(value)) => {
if value.0.is_finite() {
builder.append_value(value.0)
} else {
builder.append_value("NaN")
}
}
JsonVariantRef::String(value) => builder.append_value(*value),
JsonVariantRef::Array(values) => {
let mut list = builder.try_new_list()?;
for value in values {
append_json_variant_ref(&mut list, value)?;
}
list.finish();
}
JsonVariantRef::Object(values) => {
let mut object = builder.try_new_object()?;
for (name, value) in values {
append_json_variant_ref(&mut ObjectFieldBuilder::new(name, &mut object), value)?;
}
object.finish();
}
JsonVariantRef::Variant(value) => {
let value = decode_json_variant(value)
.map_err(|e| ArrowError::JsonError(format!("Failed to decode JSONB: {e}")))?;
append_json_value(builder, &value)?;
}
}
Ok(())
}
fn append_json_value(
builder: &mut impl VariantBuilderExt,
value: &serde_json::Value,
) -> std::result::Result<(), ArrowError> {
match value {
serde_json::Value::Null => builder.append_value(Variant::Null),
serde_json::Value::Bool(value) => builder.append_value(*value),
serde_json::Value::Number(value) => {
if let Some(value) = value.as_i64() {
builder.append_value(value);
} else if let Some(value) = value.as_u64() {
append_large_u64(builder, value)?;
} else if let Some(value) = value.as_f64() {
builder.append_value(value);
} else {
return Err(ArrowError::InvalidArgumentError(format!(
"Failed to encode JSON number as Variant: {value}"
)));
}
}
serde_json::Value::String(value) => builder.append_value(value.as_str()),
serde_json::Value::Array(values) => {
let mut list = builder.try_new_list()?;
for value in values {
append_json_value(&mut list, value)?;
}
list.finish();
}
serde_json::Value::Object(values) => {
let mut object = builder.try_new_object()?;
for (name, value) in values {
append_json_value(&mut ObjectFieldBuilder::new(name, &mut object), value)?;
}
object.finish();
}
}
Ok(())
}
/// Parquet Variant has no unsigned integer primitive. Treat u64 as i64 first, then use Decimal16
/// to represent large (larger than i64::MAX) u64.
fn append_large_u64(
builder: &mut impl VariantBuilderExt,
value: u64,
) -> std::result::Result<(), ArrowError> {
let value = VariantDecimal16::try_new(value as i128, 0).map_err(|e| {
ArrowError::InvalidArgumentError(format!(
"Failed to encode JSON large integer as Variant Decimal16: {e}"
))
})?;
builder.append_value(value);
Ok(())
}
/// Decodes an unshredded Parquet Variant array into JSON values.
pub fn variant_to_json_values(array: &ArrayRef) -> Result<Vec<Option<serde_json::Value>>> {
let variants = VariantArray::try_new(array.as_ref()).context(ArrowComputeSnafu)?;
(0..variants.len())
.map(|i| {
if variants.is_null(i) {
Ok(None)
} else {
variants
.try_value(i)
.and_then(|x| x.to_json_value())
.context(ArrowComputeSnafu)
.map(Some)
}
})
.collect()
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::*;
#[test]
fn test_variant_field_matches_canonical_layout() {
let expected = VariantArrayBuilder::new(0)
.build()
.field("remainder")
.with_nullable(true);
assert_eq!(expected, variant_field("remainder", true));
}
#[test]
fn test_variant_json_round_trip() -> Result<()> {
let values = vec![
None,
Some(serde_json::Value::Null),
Some(json!({})),
Some(json!({"nested": {"items": [1, "two", null]}})),
Some(json!({"unicode": "\u{503c}"})),
Some(json!({"max_u64": u64::MAX})),
];
let array = json_values_to_variant(&values)?;
assert_eq!(values, variant_to_json_values(&array)?);
let variants = values
.clone()
.into_iter()
.map(|x| x.map(JsonVariant::from))
.collect::<Vec<_>>();
let array = json_variants_to_variant(&variants)?;
assert_eq!(values, variant_to_json_values(&array)?);
let variants = [Some(JsonVariant::Variant(
jsonb::parse_value(br#"{"nested": true}"#).unwrap().to_vec(),
))];
let array = json_variants_to_variant(&variants)?;
assert_eq!(
vec![Some(json!({"nested": true}))],
variant_to_json_values(&array)?
);
Ok(())
}
}
+12
View File
@@ -158,6 +158,18 @@ Accounting follows the main-table write:
accepted or rejected span counts; and
- failure details are bounded before they are folded into `TraceIngestOutcome`.
A rejection detail carries the failing cause, not just a status code, so an
unusable attribute value names its column, source value, and target type. Repeats
of the same `(site, cause)` collapse into one entry with an occurrence count, and
only a fixed number of distinct entries is retained; failures past that limit are
counted but their causes are dropped, which keeps the state bounded regardless of
how many distinct bad values a request carries.
A request that rejects every span is logged at warn level. A partial success is
logged at debug level only: it repeats every export interval while the sender
keeps emitting the value, and its detail already reaches the sender through the
OTLP partial-success message.
Finally, the HTTP handler returns:
- full success when there are no rejected spans or failure details;
+152 -20
View File
@@ -25,7 +25,7 @@ use client::Output;
use common_error::ext::{BoxedError, ErrorExt};
use common_error::status_code::StatusCode;
use common_meta::rpc::ddl::TriggerReason;
use common_telemetry::warn;
use common_telemetry::{debug, warn};
use datatypes::prelude::ConcreteDataType;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
use pipeline::{GreptimePipelineParams, PipelineWay};
@@ -50,11 +50,15 @@ use crate::instance::otlp::trace_types::{
PendingTraceColumnRewrite, PreparedTraceColumnRewrites, TraceColumnRewriteError,
choose_trace_reconcile_decision, enrich_trace_reconcile_error,
is_trace_reconcile_candidate_type, prepare_trace_column_rewrites, push_observed_trace_type,
truncate_for_diagnostics,
};
use crate::metrics::{OTLP_TRACES_FAILURE_COUNT, OTLP_TRACES_ROWS};
const TRACE_FAILURE_MESSAGE_LIMIT: usize = 4;
/// Maximum characters of a failure cause echoed to the client and the log.
const TRACE_FAILURE_CAUSE_LIMIT: usize = 256;
/// Determines how trace ingestion responds to a failure before a write is dispatched.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ChunkFailureReaction {
@@ -86,7 +90,34 @@ struct TraceChunkIngestContext<'a> {
struct TraceIngestState {
aux_data: TraceAuxData,
outcome: TraceIngestOutcome,
failure_messages: Vec<String>,
failure_messages: TraceFailureMessages,
}
/// Bounded, deduplicated failure details for one trace request.
///
/// Occurrences past [`TRACE_FAILURE_MESSAGE_LIMIT`] distinct failures are
/// counted but their keys are dropped: retaining them would let this state grow
/// with the number of distinct bad values in a request.
#[derive(Debug, Default)]
struct TraceFailureMessages {
entries: Vec<TraceFailureEntry>,
suppressed_occurrences: usize,
}
#[derive(Debug)]
struct TraceFailureEntry {
label: &'static str,
/// Untruncated: two causes can agree on a truncated prefix and differ
/// exactly where the actionable detail is.
key: String,
message: String,
occurrences: usize,
}
impl TraceFailureMessages {
fn is_empty(&self) -> bool {
self.entries.is_empty()
}
}
/// How a v1 chunk should be reconciled when it is written.
@@ -863,7 +894,7 @@ impl Instance {
let mut ingest_state = TraceIngestState {
aux_data: TraceAuxData::default(),
outcome: TraceIngestOutcome::default(),
failure_messages: Vec::new(),
failure_messages: TraceFailureMessages::default(),
};
let main_result: ServerResult<()> = async {
@@ -954,6 +985,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"aux_table_update_failed",
err.status_code().as_ref(),
format!(
"Auxiliary trace tables were not fully updated ({})",
err.status_code().as_ref()
@@ -968,6 +1000,7 @@ impl Instance {
Err(err) => Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"aux_table_update_failed",
err.status_code().as_ref(),
format!(
"Auxiliary trace tables were not fully updated ({})",
err.status_code().as_ref()
@@ -985,6 +1018,35 @@ impl Instance {
ingest_state.failure_messages,
);
if let Some(error_message) = &ingest_state.outcome.error_message {
let accepted_spans = ingest_state.outcome.accepted_spans;
let rejected_spans = ingest_state.outcome.rejected_spans;
// A partial success repeats every export interval while the sender
// keeps emitting the bad value, so only a fully rejected request is
// worth a warning. Either way the detail reaches the sender: as an
// OTLP partial success, or as the status message of a 400.
//
// The detail embeds attribute keys verbatim, so it goes out as a
// Debug field; interpolating it would let a newline in an attribute
// key forge log lines.
if accepted_spans == 0 && rejected_spans > 0 {
warn!(
table_name = ingest_ctx.table_name,
rejected_spans,
error_message = ?error_message,
"OTLP trace ingest rejected every span"
);
} else {
debug!(
table_name = ingest_ctx.table_name,
accepted_spans,
rejected_spans,
error_message = ?error_message,
"OTLP trace ingest reported failures"
);
}
}
Ok(ingest_state.outcome)
}
@@ -1019,6 +1081,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating chunk write failure ({})",
err.status_code().as_ref()
@@ -1069,13 +1132,18 @@ impl Instance {
// recover valid data.
let span_count = chunks.iter().map(|chunk| chunk.rows.len()).sum::<usize>();
ingest_state.outcome.rejected_spans += span_count;
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::DiscardChunk.as_metric_label(),
// Merging discards of different sizes would make the reported
// count times its occurrences wrong.
&format!("{span_count}:{cause}"),
format!(
"Discarded {} spans after pre-write request failure ({})",
"Discarded {} spans after pre-write request failure ({}): {}",
span_count,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
return Ok(());
@@ -1110,6 +1178,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::RetryPerSpan.as_metric_label(),
"incompatible_binary_and_json",
"Chunk fallback triggered by incompatible binary and JSON values".to_string(),
);
return self
@@ -1136,6 +1205,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::RetryPerSpan.as_metric_label(),
err.status_code().as_ref(),
format!("Chunk fallback triggered by {}", err.status_code().as_ref()),
);
return self
@@ -1144,13 +1214,19 @@ impl Instance {
}
ChunkFailureReaction::DiscardChunk => {
ingest_state.outcome.rejected_spans += span_count;
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::DiscardChunk.as_metric_label(),
// Chunks discarded for one cause differ in size; merging
// them would make the reported count times its
// occurrences wrong.
&format!("{span_count}:{cause}"),
format!(
"Discarded {} spans after pre-write chunk failure ({})",
"Discarded {} spans after pre-write chunk failure ({}): {}",
span_count,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
return Ok(());
@@ -1159,6 +1235,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating pre-write chunk failure ({})",
err.status_code().as_ref()
@@ -1182,6 +1259,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating chunk write failure ({})",
err.status_code().as_ref()
@@ -1221,6 +1299,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating pre-write span failure for {}:{} ({})",
span.trace_id,
@@ -1232,14 +1311,19 @@ impl Instance {
}
ingest_state.outcome.rejected_spans += 1;
// Dedup on the cause, not the span id: one bad column rejects
// every span and would otherwise fill the entry list.
let (cause, shown_cause) = Self::trace_failure_cause(&err);
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
"span_rejected",
&cause,
format!(
"Rejected span {}:{} ({})",
"Rejected span {}:{} ({}): {}",
span.trace_id,
span.span_id,
err.status_code().as_ref()
err.status_code().as_ref(),
shown_cause
),
);
continue;
@@ -1256,6 +1340,7 @@ impl Instance {
Self::push_trace_failure_message(
&mut ingest_state.failure_messages,
ChunkFailureReaction::Propagate.as_metric_label(),
err.status_code().as_ref(),
format!(
"Propagating span write failure for {}:{} ({})",
span.trace_id,
@@ -1619,24 +1704,52 @@ impl Instance {
outcome.write_cost += cost;
}
fn push_trace_failure_message(messages: &mut Vec<String>, label: &str, message: String) {
/// Returns the full cause of a pre-write failure and its display form.
///
/// `output_msg` masks internal errors and unwraps the root cause. Dedup must
/// key on the full text: two causes can agree on a truncated prefix and
/// differ exactly where the actionable detail is.
fn trace_failure_cause(err: &error::Error) -> (String, String) {
let cause = err.output_msg();
let display = truncate_for_diagnostics(&cause, TRACE_FAILURE_CAUSE_LIMIT);
(cause, display)
}
/// Records one failure, merging repeats of `(label, key)` into a count.
fn push_trace_failure_message(
messages: &mut TraceFailureMessages,
label: &'static str,
key: &str,
message: String,
) {
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).inc();
if messages.len() < TRACE_FAILURE_MESSAGE_LIMIT {
messages.push(message);
} else if messages.len() == TRACE_FAILURE_MESSAGE_LIMIT {
tracing::debug!(
label,
limit = TRACE_FAILURE_MESSAGE_LIMIT,
"Trace ingest failure message limit reached; suppressing additional failure details"
);
if let Some(entry) = messages
.entries
.iter_mut()
.find(|entry| entry.label == label && entry.key == key)
{
entry.occurrences += 1;
return;
}
if messages.entries.len() >= TRACE_FAILURE_MESSAGE_LIMIT {
messages.suppressed_occurrences += 1;
return;
}
messages.entries.push(TraceFailureEntry {
label,
key: key.to_string(),
message,
occurrences: 1,
});
}
fn finish_trace_failure_message(
accepted_spans: usize,
rejected_spans: usize,
messages: Vec<String>,
messages: TraceFailureMessages,
) -> Option<String> {
if rejected_spans == 0 && messages.is_empty() {
return None;
@@ -1648,8 +1761,27 @@ impl Instance {
);
if !messages.is_empty() {
let details = messages
.entries
.into_iter()
.map(|entry| {
if entry.occurrences > 1 {
format!("{} (x{})", entry.message, entry.occurrences)
} else {
entry.message
}
})
.collect::<Vec<_>>()
.join("; ");
summary.push_str(": ");
summary.push_str(&messages.join("; "));
summary.push_str(&details);
}
if messages.suppressed_occurrences > 0 {
summary.push_str(&format!(
"; {} additional failures suppressed",
messages.suppressed_occurrences
));
}
Some(summary)
@@ -34,8 +34,8 @@ use servers::otlp::trace::span::{SpanEvents, SpanLinks, TraceSpan};
use servers::otlp::trace::v1::{TraceBinaryType, TraceRetryColumn};
use super::{
ChunkFailureReaction, Instance, TraceChunkRetry, TraceChunkSchemaState, TraceRequestSchema,
TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
ChunkFailureReaction, Instance, TraceChunkRetry, TraceChunkSchemaState, TraceFailureMessages,
TraceRequestSchema, TraceRequestSchemaPlan, TraceSpanMetadata, TraceTablePreAlter, chunk_owned,
wrap_trace_alter_failure,
};
use crate::metrics::OTLP_TRACES_FAILURE_COUNT;
@@ -102,22 +102,28 @@ fn test_classify_trace_prewrite_failure() {
#[test]
fn test_finish_trace_failure_message() {
let message = Instance::finish_trace_failure_message(
3,
2,
vec!["Rejected span trace:span (InvalidArguments)".to_string()],
)
.unwrap();
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
"coercion",
"Rejected span trace:span (InvalidArguments)".to_string(),
);
let message = Instance::finish_trace_failure_message(3, 2, messages).unwrap();
assert!(message.contains("Accepted 3 spans, rejected 2 spans"));
assert!(message.contains("Rejected span trace:span"));
assert_eq!(Instance::finish_trace_failure_message(2, 0, vec![]), None);
assert_eq!(
Instance::finish_trace_failure_message(2, 0, TraceFailureMessages::default()),
None
);
}
#[test]
fn test_finish_trace_failure_message_without_detail_messages() {
assert_eq!(
Instance::finish_trace_failure_message(0, 2, vec![]),
Instance::finish_trace_failure_message(0, 2, TraceFailureMessages::default()),
Some("Accepted 0 spans, rejected 2 spans".to_string())
);
}
@@ -126,39 +132,138 @@ fn test_finish_trace_failure_message_without_detail_messages() {
fn test_push_trace_failure_message_increments_labeled_counter() {
let label = "retry_per_span_counter_test";
let initial = OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get();
let mut messages = Vec::new();
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
label,
"InvalidArguments",
"Chunk fallback triggered by InvalidArguments".to_string(),
);
assert_eq!(messages.len(), 1);
assert_eq!(messages.entries.len(), 1);
assert_eq!(
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get(),
initial + 1
);
}
/// One bad column rejects every span, so repeats must collapse instead of
/// filling the bounded entry list with the same cause.
#[test]
fn test_push_trace_failure_message_collapses_repeated_cause() {
let label = "span_rejected_dedup_test";
let initial = OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get();
let mut messages = TraceFailureMessages::default();
for idx in 0..3 {
Instance::push_trace_failure_message(
&mut messages,
label,
"failed to coerce",
format!("Rejected span trace:span-{idx} (InvalidArguments): failed to coerce"),
);
}
assert_eq!(messages.entries.len(), 1);
assert_eq!(messages.entries[0].occurrences, 3);
assert_eq!(messages.suppressed_occurrences, 0);
// Every occurrence is still metered even though the details collapse.
assert_eq!(
OTLP_TRACES_FAILURE_COUNT.with_label_values(&[label]).get(),
initial + 3
);
let summary = Instance::finish_trace_failure_message(0, 3, messages).unwrap();
assert!(
summary.contains("Rejected span trace:span-0 (InvalidArguments): failed to coerce (x3)"),
"unexpected summary: {summary}"
);
}
/// Dedup identity comes from the full cause, so two failures that only differ
/// past the display limit must stay separate instead of one silently vanishing.
#[test]
fn test_trace_failure_cause_keeps_identity_beyond_the_display_limit() {
let shared_prefix = "x".repeat(400);
let (first_cause, first_shown) = Instance::trace_failure_cause(
&servers::error::InvalidParameterSnafu {
reason: format!("{shared_prefix}-int64"),
}
.build(),
);
let (second_cause, second_shown) = Instance::trace_failure_cause(
&servers::error::InvalidParameterSnafu {
reason: format!("{shared_prefix}-float64"),
}
.build(),
);
assert_eq!(first_shown, second_shown);
assert_ne!(first_cause, second_cause);
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(&mut messages, "span_rejected", &first_cause, first_shown);
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
&second_cause,
second_shown,
);
assert_eq!(messages.entries.len(), 2);
}
/// The same cause reported by two different sites must stay distinguishable.
#[test]
fn test_push_trace_failure_message_separates_labels_sharing_a_key() {
let mut messages = TraceFailureMessages::default();
Instance::push_trace_failure_message(
&mut messages,
"span_rejected",
"failed to coerce",
"Rejected span trace:span (InvalidArguments): failed to coerce".to_string(),
);
Instance::push_trace_failure_message(
&mut messages,
"discard_chunk",
"failed to coerce",
"Discarded 7 spans after pre-write chunk failure (InvalidArguments): failed to coerce"
.to_string(),
);
assert_eq!(messages.entries.len(), 2);
}
#[test]
fn test_push_trace_failure_message_caps_recorded_messages() {
let label = "retry_per_span_limit_test";
let mut messages = Vec::new();
let mut messages = TraceFailureMessages::default();
for idx in 0..=4 {
Instance::push_trace_failure_message(&mut messages, label, format!("failure-{idx}"));
for idx in 0..=5 {
Instance::push_trace_failure_message(
&mut messages,
label,
&format!("cause-{idx}"),
format!("failure-{idx}"),
);
}
assert_eq!(messages.len(), 4);
assert_eq!(
messages,
vec![
"failure-0".to_string(),
"failure-1".to_string(),
"failure-2".to_string(),
"failure-3".to_string()
]
messages
.entries
.iter()
.map(|entry| entry.message.as_str())
.collect::<Vec<_>>(),
vec!["failure-0", "failure-1", "failure-2", "failure-3"]
);
assert_eq!(messages.suppressed_occurrences, 2);
let summary = Instance::finish_trace_failure_message(0, 6, messages).unwrap();
assert!(
summary.ends_with("; 2 additional failures suppressed"),
"unexpected summary: {summary}"
);
}
+135 -18
View File
@@ -22,6 +22,39 @@ use servers::otlp::coerce::{
use crate::instance::otlp::trace_semconv::trace_semconv_fixed_type;
/// Attribute values are user data echoed back in the OTLP partial-success
/// message and the server log, so diagnostics keep at most this many characters.
const TRACE_VALUE_DIAGNOSTIC_LIMIT: usize = 16;
/// Truncates to `limit` characters, marking the cut with `...`.
///
/// Diagnostics carry user-controlled text; slicing by byte offset would panic in
/// the middle of a multi-byte sequence.
pub(super) fn truncate_for_diagnostics(text: &str, limit: usize) -> String {
match text.char_indices().nth(limit) {
Some((offset, _)) => format!("{}...", &text[..offset]),
None => text.to_string(),
}
}
/// Renders a failing trace value as `Type(value)`, e.g. `String("")`.
fn describe_trace_value(value: &ValueData, request_type: ColumnDataType) -> String {
let payload = match value {
ValueData::StringValue(string_value) => format!(
"{:?}",
truncate_for_diagnostics(string_value, TRACE_VALUE_DIAGNOSTIC_LIMIT)
),
ValueData::BoolValue(bool_value) => bool_value.to_string(),
ValueData::I64Value(int_value) => int_value.to_string(),
ValueData::F64Value(float_value) => float_value.to_string(),
ValueData::BinaryValue(bytes) => format!("{} bytes", bytes.len()),
// Other value kinds never reach trace coercion, so report the type alone.
_ => return format!("{request_type:?}"),
};
format!("{request_type:?}({payload})")
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum TraceReconcileDecision {
UseExisting(ColumnDataType),
@@ -191,8 +224,10 @@ pub(super) fn prepare_trace_column_rewrites(
let Some(value) = row.values.get(pending_rewrite.col_idx) else {
continue;
};
let Some(request_type) = value.value_data.as_ref().and_then(trace_value_datatype)
else {
let Some(request_value) = value.value_data.as_ref() else {
continue;
};
let Some(request_type) = trace_value_datatype(request_value) else {
continue;
};
if request_type == pending_rewrite.target_type {
@@ -201,20 +236,18 @@ pub(super) fn prepare_trace_column_rewrites(
let value_data =
coerce_value_data(&value.value_data, pending_rewrite.target_type, request_type)
.map_err(|_| {
TraceColumnRewriteError {
error: error::InvalidParameterSnafu {
reason: format!(
"failed to coerce trace column '{}' in table '{}' from {:?} to {:?}",
pending_rewrite.column_name,
table_name,
request_type,
pending_rewrite.target_type
),
}
.build(),
column_name: pending_rewrite.column_name.clone(),
}
.map_err(|_| TraceColumnRewriteError {
error: error::InvalidParameterSnafu {
reason: format!(
"failed to coerce trace column '{}' in table '{}' from {} to {:?}",
pending_rewrite.column_name,
table_name,
describe_trace_value(request_value, request_type),
pending_rewrite.target_type
),
}
.build(),
column_name: pending_rewrite.column_name.clone(),
})?;
values.push(PreparedTraceValueRewrite {
row_idx,
@@ -301,8 +334,8 @@ mod tests {
use super::{
PendingTraceColumnRewrite, TraceReconcileDecision, choose_trace_reconcile_decision,
enrich_trace_reconcile_error, is_trace_reconcile_candidate_type,
prepare_trace_column_rewrites, push_observed_trace_type,
describe_trace_value, enrich_trace_reconcile_error, is_trace_reconcile_candidate_type,
prepare_trace_column_rewrites, push_observed_trace_type, truncate_for_diagnostics,
};
#[test]
@@ -458,6 +491,90 @@ mod tests {
.unwrap_err();
assert_eq!(err.error.status_code(), StatusCode::InvalidArguments);
assert_eq!(err.column_name, "span_attributes.attr_int");
assert!(
err.error.to_string().contains(
"failed to coerce trace column 'span_attributes.attr_int' in table \
'trace_type_atomicity' from String(\"not_a_number\") to Int64"
),
"unexpected error message: {}",
err.error
);
}
/// The PHP instrumentation case: an empty string must be distinguishable
/// from any other unparsable value in the reported diagnostics.
#[test]
fn test_prepare_trace_column_rewrites_reports_empty_string_value() {
let rows = vec![Row {
values: vec![Value {
value_data: Some(ValueData::StringValue(String::new())),
}],
}];
let pending_rewrites = vec![PendingTraceColumnRewrite {
col_idx: 0,
target_type: ColumnDataType::Int64,
column_name: "span_attributes.http.response.body.size".to_string(),
}];
let err = prepare_trace_column_rewrites(&rows, pending_rewrites, "opentelemetry_traces")
.unwrap_err();
assert!(
err.error.to_string().contains(
"'span_attributes.http.response.body.size' in table 'opentelemetry_traces' \
from String(\"\") to Int64"
),
"unexpected error message: {}",
err.error
);
}
#[test]
fn test_describe_trace_value_bounds_and_escapes_strings() {
assert_eq!(
describe_trace_value(
&ValueData::StringValue(String::new()),
ColumnDataType::String
),
r#"String("")"#
);
assert_eq!(
describe_trace_value(
&ValueData::StringValue("a\tb\"c".to_string()),
ColumnDataType::String
),
r#"String("a\tb\"c")"#
);
assert_eq!(
describe_trace_value(
&ValueData::StringValue("0123456789abcdefghij".to_string()),
ColumnDataType::String
),
r#"String("0123456789abcdef...")"#
);
}
#[test]
fn test_describe_trace_value_omits_binary_content() {
assert_eq!(
describe_trace_value(
&ValueData::BinaryValue(vec![1_u8, 2, 3]),
ColumnDataType::Binary
),
"Binary(3 bytes)"
);
assert_eq!(
describe_trace_value(&ValueData::F64Value(1.5), ColumnDataType::Float64),
"Float64(1.5)"
);
}
/// Truncation runs over user-supplied text, so it must not split a
/// multi-byte character.
#[test]
fn test_truncate_for_diagnostics_cuts_on_char_boundary() {
assert_eq!(truncate_for_diagnostics("日本語テキスト", 3), "日本語...");
assert_eq!(truncate_for_diagnostics("short", 16), "short");
assert_eq!(truncate_for_diagnostics("exact", 5), "exact");
}
#[test]
+1 -9
View File
@@ -47,7 +47,7 @@ use servers::http::header::{CONTENT_ENCODING_SNAPPY, CONTENT_TYPE_PROTOBUF, coll
use servers::http::prom_store::PHYSICAL_TABLE_PARAM;
use servers::interceptor::{PromStoreProtocolInterceptor, PromStoreProtocolInterceptorRef};
use servers::pending_rows_batcher::PendingRowsSchemaAlterer;
use servers::prom_store::{self, Metrics};
use servers::prom_store;
use servers::query_handler::{
PromStoreProtocolHandler, PromStoreProtocolHandlerRef, PromStoreResponse,
};
@@ -648,10 +648,6 @@ impl PromStoreProtocolHandler for Instance {
.fail(),
}
}
async fn ingest_metrics(&self, _metrics: Metrics) -> ServerResult<()> {
todo!();
}
}
impl Instance {
@@ -791,10 +787,6 @@ impl PromStoreProtocolHandler for ExportMetricHandler {
) -> ServerResult<PromStoreResponse> {
unreachable!();
}
async fn ingest_metrics(&self, _metrics: Metrics) -> ServerResult<()> {
unreachable!();
}
}
#[cfg(test)]
@@ -13,6 +13,7 @@
// limitations under the License.
use std::cmp::Ordering;
use std::sync::Arc;
use api::v1::meta::{HeartbeatRequest, Role};
use common_meta::datanode::{DatanodeStatKey, DatanodeStatValue, Stat};
@@ -24,6 +25,7 @@ use common_meta::rpc::store::PutRequest;
use common_telemetry::{error, info, warn};
use dashmap::DashMap;
use snafu::ResultExt;
use tokio::sync::Mutex;
use crate::error::{self, Result};
use crate::handler::{HandleControl, HeartbeatAccumulator, HeartbeatHandler};
@@ -70,7 +72,7 @@ impl EpochStats {
const DEFAULT_FLUSH_STATS_FACTOR: usize = 3;
pub struct CollectStatsHandler {
stats_cache: DashMap<DatanodeStatKey, EpochStats>,
stats_cache: DashMap<DatanodeStatKey, Arc<Mutex<EpochStats>>>,
flush_stats_factor: usize,
}
@@ -106,10 +108,16 @@ impl HeartbeatHandler for CollectStatsHandler {
};
let key = current_stat.stat_key();
let mut entry = self.stats_cache.entry(key).or_default();
let state = {
let entry = self
.stats_cache
.entry(key)
.or_insert_with(|| Arc::new(Mutex::new(EpochStats::default())));
Arc::clone(entry.value())
};
let mut epoch_stats = state.lock().await;
let key: Vec<u8> = key.into();
let epoch_stats = entry.value_mut();
let refresh = if let Some(epoch) = epoch_stats.epoch() {
match current_stat.node_epoch.cmp(&epoch) {
@@ -170,7 +178,7 @@ impl HeartbeatHandler for CollectStatsHandler {
}
}
async fn rewrite_node_address(ctx: &mut Context, stat: &Stat) {
async fn rewrite_node_address(ctx: &Context, stat: &Stat) {
let peer = Peer {
id: stat.id,
addr: stat.addr.clone(),
@@ -217,10 +225,188 @@ async fn rewrite_node_address(ctx: &mut Context, stat: &Stat) {
#[cfg(test)]
mod tests {
use std::any::Any;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex as StdMutex, mpsc};
use std::thread;
use std::time::Duration;
use common_meta::datanode::DatanodeStatKey;
use common_meta::error::{Error as MetaError, Result as MetaResult};
use common_meta::kv_backend::{KvBackend, KvBackendRef, ResettableKvBackend, TxnService};
use common_meta::rpc::store::{
BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse,
BatchPutRequest, BatchPutResponse, DeleteRangeRequest, DeleteRangeResponse, PutResponse,
RangeRequest, RangeResponse,
};
use tokio::sync::Semaphore;
use tokio::time::{sleep, timeout};
use super::*;
use crate::handler::test_utils::TestEnv;
use crate::service::store::cached_kv::LeaderCachedKvBackend;
struct ControlledKvBackend {
recorded_puts: StdMutex<Vec<PutRequest>>,
put_entered: Semaphore,
put_release: Semaphore,
block_next_put: AtomicBool,
delay_puts: AtomicBool,
active_puts: AtomicUsize,
max_active_puts: AtomicUsize,
}
impl ControlledKvBackend {
fn new() -> Self {
Self {
recorded_puts: StdMutex::new(Vec::new()),
put_entered: Semaphore::new(0),
put_release: Semaphore::new(0),
block_next_put: AtomicBool::new(false),
delay_puts: AtomicBool::new(false),
active_puts: AtomicUsize::new(0),
max_active_puts: AtomicUsize::new(0),
}
}
fn block_next_put(&self) {
self.block_next_put.store(true, AtomicOrdering::Relaxed);
}
async fn wait_for_blocked_put(&self) {
self.put_entered.acquire().await.unwrap().forget();
}
fn release_one_put(&self) {
self.put_release.add_permits(1);
}
fn set_delay_puts(&self, delay: bool) {
self.delay_puts.store(delay, AtomicOrdering::Relaxed);
}
fn clear_recorded_puts(&self) {
self.recorded_puts.lock().unwrap().clear();
}
fn recorded_puts(&self) -> Vec<PutRequest> {
self.recorded_puts.lock().unwrap().clone()
}
fn max_active_puts(&self) -> usize {
self.max_active_puts.load(AtomicOrdering::Relaxed)
}
fn start_put(&self, req: &PutRequest) -> PutGuard<'_> {
self.recorded_puts.lock().unwrap().push(req.clone());
let active = self.active_puts.fetch_add(1, AtomicOrdering::Relaxed) + 1;
self.max_active_puts
.fetch_max(active, AtomicOrdering::Relaxed);
PutGuard { backend: self }
}
}
struct PutGuard<'a> {
backend: &'a ControlledKvBackend,
}
impl Drop for PutGuard<'_> {
fn drop(&mut self) {
self.backend
.active_puts
.fetch_sub(1, AtomicOrdering::Relaxed);
}
}
#[async_trait::async_trait]
impl TxnService for ControlledKvBackend {
type Error = MetaError;
}
#[async_trait::async_trait]
impl KvBackend for ControlledKvBackend {
fn name(&self) -> &str {
"controlled"
}
fn as_any(&self) -> &dyn Any {
self
}
async fn range(&self, _req: RangeRequest) -> MetaResult<RangeResponse> {
unimplemented!()
}
async fn put(&self, req: PutRequest) -> MetaResult<PutResponse> {
let _guard = self.start_put(&req);
if self.block_next_put.swap(false, AtomicOrdering::Relaxed) {
self.put_entered.add_permits(1);
self.put_release.acquire().await.unwrap().forget();
}
if self.delay_puts.load(AtomicOrdering::Relaxed) {
sleep(Duration::from_millis(10)).await;
}
Ok(PutResponse::default())
}
async fn batch_put(&self, _req: BatchPutRequest) -> MetaResult<BatchPutResponse> {
unimplemented!()
}
async fn batch_get(&self, _req: BatchGetRequest) -> MetaResult<BatchGetResponse> {
unimplemented!()
}
async fn delete_range(&self, _req: DeleteRangeRequest) -> MetaResult<DeleteRangeResponse> {
unimplemented!()
}
async fn batch_delete(&self, _req: BatchDeleteRequest) -> MetaResult<BatchDeleteResponse> {
unimplemented!()
}
}
impl ResettableKvBackend for ControlledKvBackend {
fn reset(&self) {
self.clear_recorded_puts();
}
fn as_kv_backend_ref(self: Arc<Self>) -> KvBackendRef {
self
}
}
fn stat(node_id: u64, epoch: u64, marker: u64, addr: &str) -> Stat {
Stat {
timestamp_millis: marker as i64,
id: node_id,
addr: addr.to_string(),
region_num: marker,
node_epoch: epoch,
..Default::default()
}
}
async fn handle_stat(
handler: Arc<CollectStatsHandler>,
mut ctx: Context,
stat: Stat,
) -> Result<HandleControl> {
let mut acc = HeartbeatAccumulator {
stat: Some(stat),
..Default::default()
};
handler
.handle(&HeartbeatRequest::default(), &mut ctx, &mut acc)
.await
}
fn use_controlled_address_backend(ctx: &mut Context) -> Arc<ControlledKvBackend> {
let backend = Arc::new(ControlledKvBackend::new());
ctx.leader_cached_kv_backend =
Arc::new(LeaderCachedKvBackend::with_always_leader(backend.clone()));
backend
}
#[tokio::test]
async fn test_handle_datanode_stats() {
@@ -250,6 +436,105 @@ mod tests {
assert_eq!(handler.flush_stats_factor, val.stats.len());
}
#[test]
fn test_same_datanode_wait_keeps_current_thread_runtime_responsive() {
let (backend_tx, backend_rx) = mpsc::sync_channel(1);
let (timer_tx, timer_rx) = mpsc::sync_channel(1);
let (done_tx, done_rx) = mpsc::sync_channel(1);
let worker = thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_time()
.build()
.unwrap()
.block_on(async move {
let env = TestEnv::new();
let mut ctx = env.ctx();
let address_backend = use_controlled_address_backend(&mut ctx);
address_backend.block_next_put();
backend_tx.send(address_backend.clone()).unwrap();
let handler = Arc::new(CollectStatsHandler::default());
let first = tokio::spawn(handle_stat(
handler.clone(),
ctx.clone(),
stat(101, 1, 1, "dn-101-v1"),
));
address_backend.wait_for_blocked_put().await;
let second =
tokio::spawn(handle_stat(handler, ctx, stat(101, 1, 2, "dn-101-v1")));
tokio::spawn(async move {
sleep(Duration::from_millis(10)).await;
timer_tx.send(()).unwrap();
});
tokio::task::yield_now().await;
first.await.unwrap().unwrap();
second.await.unwrap().unwrap();
done_tx.send(()).unwrap();
});
});
let address_backend = backend_rx.recv_timeout(Duration::from_secs(1)).unwrap();
let timer_result = timer_rx.recv_timeout(Duration::from_secs(1));
address_backend.release_one_put();
timer_result.expect("the current-thread runtime must remain responsive");
done_rx
.recv_timeout(Duration::from_secs(1))
.expect("both heartbeat handlers must complete after releasing the write");
worker.join().unwrap();
}
#[tokio::test]
async fn test_concurrent_flush_persists_every_stat_once() {
let env = TestEnv::new();
let mut ctx = env.ctx();
let stats_backend = Arc::new(ControlledKvBackend::new());
ctx.in_memory = stats_backend.clone();
stats_backend.set_delay_puts(true);
let flush_stats_factor = 3;
let handler = Arc::new(CollectStatsHandler::new(Some(flush_stats_factor)));
handle_stat(handler.clone(), ctx.clone(), stat(101, 1, 0, "dn-101"))
.await
.unwrap();
stats_backend.clear_recorded_puts();
let mut tasks = Vec::with_capacity(2 * flush_stats_factor);
for marker in 1..=(2 * flush_stats_factor) {
tasks.push(tokio::spawn(handle_stat(
handler.clone(),
ctx.clone(),
stat(101, 1, marker as u64, "dn-101"),
)));
}
for task in tasks {
timeout(Duration::from_secs(1), task)
.await
.unwrap()
.unwrap()
.unwrap();
}
let puts = stats_backend.recorded_puts();
assert_eq!(2, puts.len());
let mut markers = puts
.into_iter()
.flat_map(|put| {
let value: DatanodeStatValue = put.value.try_into().unwrap();
value
.stats
.into_iter()
.map(|stat| stat.region_num)
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
markers.sort_unstable();
assert_eq!((1..=6).collect::<Vec<_>>(), markers);
assert_eq!(1, stats_backend.max_active_puts());
}
async fn handle_request_many_times(
mut ctx: Context,
handler: &CollectStatsHandler,
-1
View File
@@ -87,7 +87,6 @@ mysql_common = "0.38"
notify.workspace = true
object-pool = "0.5"
once_cell.workspace = true
openmetrics-parser = "0.4"
# Wait for https://github.com/databendlabs/opensrv/pull/81
opensrv-mysql = { git = "https://github.com/GreptimeTeam/opensrv", rev = "6c5a451544194b7bb60a8318d155d4f892b49f2c" }
opentelemetry-proto.workspace = true
+1 -1
View File
@@ -1 +1 @@
v0.13.12
v0.13.13
-9
View File
@@ -770,7 +770,6 @@ mod tests {
use super::*;
use crate::prom_remote_write::validation::PromValidationMode;
use crate::prom_store::Metrics;
use crate::query_handler::PromStoreProtocolHandler;
#[test]
@@ -966,10 +965,6 @@ mod tests {
) -> Result<PromStoreResponse> {
unimplemented!()
}
async fn ingest_metrics(&self, _metrics: Metrics) -> Result<()> {
unimplemented!()
}
}
#[tokio::test]
@@ -1067,9 +1062,5 @@ mod tests {
) -> Result<PromStoreResponse> {
unimplemented!()
}
async fn ingest_metrics(&self, _metrics: Metrics) -> Result<()> {
unimplemented!()
}
}
}
-6
View File
@@ -34,7 +34,6 @@ use datafusion::dataframe::DataFrame;
use datafusion::prelude::{Expr, col, lit, regexp_match};
use datafusion_common::ScalarValue;
use datafusion_expr::LogicalPlan;
use openmetrics_parser::{MetricsExposition, PrometheusType, PrometheusValue};
use snafu::{OptionExt, ResultExt, ensure};
use snap::raw::{Decoder, Encoder};
@@ -95,11 +94,6 @@ pub fn is_physical_table_selection_label(label: &str) -> bool {
label == PHYSICAL_TABLE_LABEL || label == PHYSICAL_TABLE_LABEL_ALT
}
/// Metrics for push gateway protocol
pub struct Metrics {
pub exposition: MetricsExposition<PrometheusType, PrometheusValue>,
}
/// Get table name from remote query
pub fn table_name(q: &Query) -> Result<String> {
let mut matchers = q
-3
View File
@@ -54,7 +54,6 @@ use crate::error::Result;
use crate::http::jaeger::QueryTraceParams;
use crate::influxdb::InfluxdbRequest;
use crate::opentsdb::codec::DataPoint;
use crate::prom_store::Metrics;
pub type OpentsdbProtocolHandlerRef = Arc<dyn OpentsdbProtocolHandler + Send + Sync>;
pub type InfluxdbLineProtocolHandlerRef = Arc<dyn InfluxdbLineProtocolHandler + Send + Sync>;
pub type PromStoreProtocolHandlerRef = Arc<dyn PromStoreProtocolHandler + Send + Sync>;
@@ -125,8 +124,6 @@ pub trait PromStoreProtocolHandler {
/// Handling prometheus remote read requests
async fn read(&self, request: ReadRequest, ctx: QueryContextRef) -> Result<PromStoreResponse>;
/// Handling push gateway requests
async fn ingest_metrics(&self, metrics: Metrics) -> Result<()>;
}
/// Outcome of an OTLP metrics ingestion: the main write output plus an
+1 -5
View File
@@ -44,7 +44,7 @@ use servers::http::{HttpOptions, HttpServerBuilder};
use servers::prom_remote_write::v2::test_util as remote_write_v2;
use servers::prom_remote_write::validation::PromValidationMode;
use servers::prom_store;
use servers::prom_store::{Metrics, snappy_compress};
use servers::prom_store::snappy_compress;
use servers::query_handler::sql::SqlQueryHandler;
use servers::query_handler::{PromStoreProtocolHandler, PromStoreResponse};
use session::context::QueryContextRef;
@@ -146,10 +146,6 @@ impl PromStoreProtocolHandler for DummyInstance {
body: response.encode_to_vec(),
})
}
async fn ingest_metrics(&self, _metrics: Metrics) -> Result<()> {
unimplemented!();
}
}
#[async_trait]
+10 -1
View File
@@ -339,6 +339,13 @@ pub enum Error {
#[snafu(implicit)]
location: Location,
},
#[snafu(transparent)]
Datatypes {
#[snafu(implicit)]
location: Location,
source: datatypes::error::Error,
},
}
impl ErrorExt for Error {
@@ -384,7 +391,9 @@ impl ErrorExt for Error {
#[cfg(feature = "enterprise")]
InvalidTriggerWebhookOption { .. } => StatusCode::InvalidArguments,
SerializeColumnDefaultConstraint { source, .. } => source.status_code(),
SerializeColumnDefaultConstraint { source, .. } | Datatypes { source, .. } => {
source.status_code()
}
ConvertToGrpcDataType { source, .. } => source.status_code(),
SqlCommon { source, .. } => source.status_code(),
+2 -2
View File
@@ -711,8 +711,8 @@ impl<'a> ParserContext<'a> {
let mut extensions = ColumnExtensions::default();
let data_type =
if let Some((data_type, type_hints)) = json::parse_json2_type_and_hints(parser)? {
extensions.json_type_hints = type_hints;
if let Some((data_type, options)) = json::parse_json2_type_and_options(parser)? {
extensions.json2_options = options;
data_type
} else {
parser.parse_data_type().context(SyntaxSnafu)?
+134 -20
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use datatypes::extension::json::JSON2_REMAINDER_FIELD_NAME;
use datatypes::json::JSON2_MAX_STRUCTURED_DEPTH;
use snafu::{ResultExt, ensure};
use sqlparser::ast::{DataType, ExactNumberInfo, Expr, ObjectName, UnaryOperator};
@@ -22,14 +23,15 @@ use sqlparser::tokenizer::Token;
use crate::ast::Ident;
use crate::error::{InvalidSqlSnafu, Result, SyntaxSnafu};
use crate::parsers::create_parser::{INVERTED, SKIPPING};
use crate::statements::create::JsonTypeHint;
use crate::statements::create::{Json2Options, JsonTypeHint};
use crate::statements::transform::type_alias::get_type_by_alias;
const JSON2_TYPE_NAME: &str = "JSON2";
const MAX_AUTO_EXPANDED_PATHS: &str = "max_auto_expanded_paths";
pub(super) fn parse_json2_type_and_hints(
pub(super) fn parse_json2_type_and_options(
parser: &mut Parser<'_>,
) -> Result<Option<(DataType, Vec<JsonTypeHint>)>> {
) -> Result<Option<(DataType, Option<Json2Options>)>> {
let token = parser.peek_token();
let Token::Word(word) = &token.token else {
return Ok(None);
@@ -41,26 +43,62 @@ pub(super) fn parse_json2_type_and_hints(
parser.next_token();
let data_type = DataType::Custom(ObjectName::from(vec![Ident::new(JSON2_TYPE_NAME)]), vec![]);
let type_hints = if parser.consume_token(&Token::LParen) {
parse_json2_type_hints(parser)?
let options = if parser.consume_token(&Token::LParen) {
parse_json2_options(parser)?
} else {
vec![]
None
};
Ok(Some((data_type, type_hints)))
Ok(Some((data_type, options)))
}
fn parse_json2_type_hints(parser: &mut Parser<'_>) -> Result<Vec<JsonTypeHint>> {
let mut hints = Vec::new();
fn parse_json2_options(parser: &mut Parser<'_>) -> Result<Option<Json2Options>> {
if parser.consume_token(&Token::RParen) {
return Ok(hints);
return Ok(None);
}
let mut max_auto_expanded_paths = None;
let mut type_hints = Vec::new();
loop {
let hint = parse_json2_type_hint(parser)?;
ensure_no_path_conflict(&hints, &hint.path)?;
hints.push(hint);
let token = parser.peek_token();
let is_max_auto_expanded_paths = matches!(
&token.token,
Token::Word(word)
if word.quote_style.is_none()
&& word.value.eq_ignore_ascii_case(MAX_AUTO_EXPANDED_PATHS)
);
if is_max_auto_expanded_paths {
parser.next_token();
ensure!(
max_auto_expanded_paths.is_none(),
InvalidSqlSnafu {
msg: format!("duplicated JSON2 option '{MAX_AUTO_EXPANDED_PATHS}'")
}
);
parser.expect_token(&Token::Eq).context(SyntaxSnafu)?;
let token = parser.next_token();
let Token::Number(value, _) = token.token else {
return InvalidSqlSnafu {
msg: format!(
"JSON2 option '{MAX_AUTO_EXPANDED_PATHS}' expects a non-negative integer"
),
}
.fail();
};
max_auto_expanded_paths = Some(value.parse::<u32>().map_err(|_| {
InvalidSqlSnafu {
msg: format!(
"JSON2 option '{MAX_AUTO_EXPANDED_PATHS}' expects a non-negative integer"
),
}
.build()
})?);
} else {
let hint = parse_json2_type_hint(parser)?;
ensure_no_path_conflict(&type_hints, &hint.path)?;
type_hints.push(hint);
}
if parser.consume_token(&Token::Comma) {
if parser.consume_token(&Token::RParen) {
@@ -72,11 +110,22 @@ fn parse_json2_type_hints(parser: &mut Parser<'_>) -> Result<Vec<JsonTypeHint>>
}
}
Ok(hints)
Ok(Some(Json2Options {
max_auto_expanded_paths,
type_hints,
}))
}
fn parse_json2_type_hint(parser: &mut Parser<'_>) -> Result<JsonTypeHint> {
let path = parse_json2_path(parser)?;
ensure!(
path.first().is_none_or(|x| x != JSON2_REMAINDER_FIELD_NAME),
InvalidSqlSnafu {
msg: format!(
"JSON2 type hint path cannot be rooted at reserved field '{JSON2_REMAINDER_FIELD_NAME}'"
)
}
);
ensure!(
path.len() <= JSON2_MAX_STRUCTURED_DEPTH,
InvalidSqlSnafu {
@@ -309,7 +358,7 @@ CREATE TABLE traces (
column.column_def.data_type,
DataType::Custom(_, _)
));
let hints = column.extensions.json_type_hints;
let hints = column.extensions.json2_options.unwrap().type_hints;
assert_eq!(hints.len(), 4);
assert_eq!(hints[0].path, vec!["service.name"]);
@@ -339,6 +388,71 @@ CREATE TABLE traces (
assert!(hints[3].nullable);
}
#[test]
fn test_parse_json2_max_auto_expanded_paths() {
let column = parse_json2_column(
r#"
CREATE TABLE traces (
log_json_data JSON2 (
http.method STRING,
max_auto_expanded_paths = 0
),
ts TIMESTAMP TIME INDEX,
)"#,
);
let options = column.extensions.json2_options.unwrap();
assert_eq!(options.max_auto_expanded_paths, Some(0));
assert_eq!(options.type_hints.len(), 1);
let empty = parse_json2_column(
r#"
CREATE TABLE traces (
log_json_data JSON2 (),
ts TIMESTAMP TIME INDEX,
)"#,
);
assert!(empty.extensions.json2_options.is_none());
let quoted = parse_json2_column(
r#"
CREATE TABLE traces (
log_json_data JSON2 (
"max_auto_expanded_paths" STRING,
nested."!__remainder__!" STRING
),
ts TIMESTAMP TIME INDEX,
)"#,
);
let options = quoted.extensions.json2_options.unwrap();
assert_eq!(options.max_auto_expanded_paths, None);
assert_eq!(options.type_hints.len(), 2);
}
#[test]
fn test_parse_json2_max_auto_expanded_paths_rejects_invalid_options() {
for options in [
"max_auto_expanded_paths = 0, max_auto_expanded_paths = 1",
"max_auto_expanded_paths = -1",
"max_auto_expanded_paths = 1.5",
"max_auto_expanded_paths = 4294967296",
r#""!__remainder__!".value STRING"#,
] {
let sql = format!(
"CREATE TABLE traces (log_json_data JSON2 ({options}), ts TIMESTAMP TIME INDEX)"
);
assert!(
ParserContext::create_with_dialect(
&sql,
&GreptimeDbDialect {},
ParseOptions::default()
)
.is_err(),
"{options}"
);
}
}
#[test]
fn test_parse_json2_type_hint_default_nullable() {
let column = parse_json2_column(
@@ -349,7 +463,7 @@ CREATE TABLE traces (
)"#,
);
let hints = column.extensions.json_type_hints;
let hints = column.extensions.json2_options.unwrap().type_hints;
assert_eq!(hints.len(), 1);
assert!(hints[0].nullable);
}
@@ -369,7 +483,7 @@ CREATE TABLE traces (
)"#,
);
let hints = column.extensions.json_type_hints;
let hints = column.extensions.json2_options.unwrap().type_hints;
assert_eq!(hints.len(), 4);
assert_eq!(hints[0].path, vec!["a", "b"]);
assert_eq!(hints[1].path, vec!["x", "y"]);
@@ -402,7 +516,7 @@ CREATE TABLE traces (
)"#,
);
let hints = column.extensions.json_type_hints;
let hints = column.extensions.json2_options.unwrap().type_hints;
assert_eq!(hints.len(), 14);
for hint in hints.iter().take(6) {
assert_eq!(hint.data_type, DataType::BigInt(None));
@@ -428,7 +542,7 @@ CREATE TABLE traces (
)"#,
);
let hints = column.extensions.json_type_hints;
let hints = column.extensions.json2_options.unwrap().type_hints;
assert_eq!(hints.len(), 2);
assert_eq!(
hints[0]
+3 -15
View File
@@ -718,11 +718,7 @@ mod tests {
"true".to_string(),
),
])),
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: None,
..Default::default()
},
};
@@ -749,17 +745,13 @@ mod tests {
options: vec![],
},
extensions: ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::from([
("metric".to_string(), "cosine".to_string()),
("connectivity".to_string(), "32".to_string()),
("expansion_add".to_string(), "200".to_string()),
("expansion_search".to_string(), "100".to_string()),
])),
..Default::default()
},
};
@@ -790,12 +782,8 @@ mod tests {
options: vec![],
},
extensions: ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::default()),
..Default::default()
},
};
+80 -64
View File
@@ -136,7 +136,28 @@ pub struct ColumnExtensions {
pub inverted_index_options: Option<OptionMap>,
/// Vector index options for HNSW-based vector similarity search.
pub vector_index_options: Option<OptionMap>,
pub json_type_hints: Vec<JsonTypeHint>,
/// JSON2-specific column options.
pub json2_options: Option<Json2Options>,
}
/// JSON2-specific options represented in the SQL AST.
#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
pub struct Json2Options {
/// Maximum number of unhinted JSON2 paths expanded into Arrow fields.
pub(crate) max_auto_expanded_paths: Option<u32>,
/// Paths stored as explicitly typed JSON2 fields.
pub(crate) type_hints: Vec<JsonTypeHint>,
}
impl Display for Json2Options {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
let mut options = Vec::with_capacity(self.type_hints.len() + 1);
if let Some(max) = self.max_auto_expanded_paths {
options.push(format!("max_auto_expanded_paths = {max}"));
}
options.extend(self.type_hints.iter().map(format_json_type_hint));
write!(f, "(\n {}\n )", options.iter().join(",\n "))
}
}
#[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)]
@@ -180,12 +201,8 @@ impl Display for Column {
}
write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?;
if !self.extensions.json_type_hints.is_empty() {
write!(
f,
"{}",
format_json_type_hints(&self.extensions.json_type_hints)
)?;
if let Some(options) = &self.extensions.json2_options {
write!(f, "{options}")?;
}
for option in &self.column_def.options {
write!(f, " {option}")?;
@@ -335,29 +352,30 @@ impl ColumnExtensions {
}
pub fn build_json_settings(&self) -> Result<Option<JsonSettings>> {
if self.json_type_hints.is_empty() {
let Some(options) = &self.json2_options else {
return Ok(None);
}
};
Ok(Some(JsonSettings::new(
self.json_type_hints
.iter()
.map(|hint| {
Ok(datatypes::json::JsonTypeHint {
path: hint.path.clone(),
data_type: json_type_hint_concrete_data_type(&hint.data_type)?,
nullable: hint.nullable,
default_constraint: build_json_type_hint_default_constraint(hint)?,
inverted_index: hint.inverted_index,
})
let type_hints = options
.type_hints
.iter()
.map(|hint| {
Ok(datatypes::json::JsonTypeHint {
path: hint.path.clone(),
data_type: json_type_hint_concrete_data_type(&hint.data_type)?,
nullable: hint.nullable,
default_constraint: build_json_type_hint_default_constraint(hint)?,
inverted_index: hint.inverted_index,
})
.collect::<Result<Vec<_>>>()?,
)))
})
.collect::<Result<Vec<_>>>()?;
let settings = JsonSettings::try_new(type_hints, options.max_auto_expanded_paths)?;
Ok(Some(settings))
}
pub fn set_json_settings(&mut self, settings: JsonSettings) -> Result<()> {
self.json_type_hints = settings
.type_hints
let (type_hints, max_auto_expanded_paths) = settings.into_parts();
let type_hints = type_hints
.into_iter()
.map(|hint| {
let data_type = json_type_hint_sql_data_type(&hint.data_type)?;
@@ -374,6 +392,11 @@ impl ColumnExtensions {
})
})
.collect::<Result<Vec<_>>>()?;
self.json2_options = (max_auto_expanded_paths.is_some() || !type_hints.is_empty())
.then_some(Json2Options {
max_auto_expanded_paths,
type_hints,
});
Ok(())
}
}
@@ -487,13 +510,6 @@ fn format_json_type_hint(hint: &JsonTypeHint) -> String {
)
}
fn format_json_type_hints(hints: &[JsonTypeHint]) -> String {
format!(
"(\n {}\n )",
hints.iter().map(format_json_type_hint).join(",\n ")
)
}
fn format_json_path_segment(segment: &str) -> String {
format!("\"{}\"", segment.replace('"', "\"\""))
}
@@ -786,6 +802,7 @@ mod tests {
use datatypes::schema::ColumnDefaultConstraint;
use datatypes::value::Value;
use super::*;
use crate::dialect::GreptimeDbDialect;
use crate::error::Error;
use crate::parser::{ParseOptions, ParserContext};
@@ -1066,7 +1083,7 @@ ENGINE=mito
.build_json_settings()
.unwrap()
.unwrap();
let hints = settings.type_hints;
let hints = settings.type_hints();
assert_eq!(hints[0].data_type, ConcreteDataType::int64_datatype());
assert_eq!(
@@ -1118,10 +1135,10 @@ ENGINE=mito
}
#[test]
fn test_set_json_settings_normalizes_type_hint_sql_types() {
fn test_set_json_settings_normalizes_type_hint_sql_types() -> Result<()> {
let mut extensions = super::ColumnExtensions::default();
extensions
.set_json_settings(JsonSettings::new(vec![
let settings = JsonSettings::try_new(
vec![
DatatypeJsonTypeHint {
path: vec!["i".to_string()],
data_type: ConcreteDataType::int32_datatype(),
@@ -1157,36 +1174,51 @@ ENGINE=mito
default_constraint: None,
inverted_index: false,
},
]))
.unwrap();
],
None,
)?;
extensions.set_json_settings(settings)?;
assert_eq!(
extensions
.json_type_hints
.json2_options
.unwrap()
.type_hints
.iter()
.map(|hint| hint.data_type.to_string())
.collect::<Vec<_>>(),
vec!["BIGINT", "DOUBLE", "BIGINT UNSIGNED", "STRING", "BOOLEAN"]
);
Ok(())
}
#[test]
fn test_set_json_settings_rejects_unsupported_type_hint_type() {
let mut extensions = super::ColumnExtensions::default();
let err = extensions
.set_json_settings(JsonSettings::new(vec![DatatypeJsonTypeHint {
fn test_set_json_settings_rejects_unsupported_type_hint_type() -> Result<()> {
let err = JsonSettings::try_new(
vec![DatatypeJsonTypeHint {
path: vec!["u".to_string()],
data_type: ConcreteDataType::date_datatype(),
nullable: true,
default_constraint: None,
inverted_index: false,
}]))
.unwrap_err();
}],
None,
)
.unwrap_err();
assert!(
err.to_string()
.contains("unsupported JSON2 type hint data type")
);
Ok(())
}
#[test]
fn test_set_empty_json_settings_omits_json2_options() -> Result<()> {
let mut extensions = ColumnExtensions::default();
extensions.set_json_settings(JsonSettings::default())?;
assert!(extensions.json2_options.is_none());
Ok(())
}
#[test]
@@ -1364,15 +1396,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
// Test zero connectivity should fail
let extensions = ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::from([(
"connectivity".to_string(),
"0".to_string(),
)])),
..Default::default()
};
let result = extensions.build_vector_index_options();
assert!(result.is_err());
@@ -1385,15 +1413,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
// Test zero expansion_add should fail
let extensions = ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::from([(
"expansion_add".to_string(),
"0".to_string(),
)])),
..Default::default()
};
let result = extensions.build_vector_index_options();
assert!(result.is_err());
@@ -1406,15 +1430,11 @@ AS SELECT number FROM numbers_input where number > 10"#,
// Test zero expansion_search should fail
let extensions = ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::from([(
"expansion_search".to_string(),
"0".to_string(),
)])),
..Default::default()
};
let result = extensions.build_vector_index_options();
assert!(result.is_err());
@@ -1427,16 +1447,12 @@ AS SELECT number FROM numbers_input where number > 10"#,
// Test valid values should succeed
let extensions = ColumnExtensions {
fulltext_index_options: None,
vector_options: None,
skipping_index_options: None,
inverted_index_options: None,
json_type_hints: vec![],
vector_index_options: Some(OptionMap::from([
("connectivity".to_string(), "32".to_string()),
("expansion_add".to_string(), "200".to_string()),
("expansion_search".to_string(), "100".to_string()),
])),
..Default::default()
};
let result = extensions.build_vector_index_options();
assert!(result.is_ok());
+139
View File
@@ -0,0 +1,139 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use common_query::Output;
use frontend::instance::Instance;
use servers::query_handler::sql::SqlQueryHandler;
use session::context::QueryContext;
use tests_integration::standalone::GreptimeDbStandaloneBuilder;
async fn execute_sql(instance: &Instance, sql: &str) -> Output {
SqlQueryHandler::do_query(instance, sql, QueryContext::arc())
.await
.remove(0)
.unwrap()
}
#[tokio::test(flavor = "multi_thread")]
async fn test_dict_groupby_sst() {
common_telemetry::init_default_ut_logging();
let standalone = GreptimeDbStandaloneBuilder::new("dict_groupby_sst")
.build()
.await;
let frontend = standalone.fe_instance();
execute_sql(
frontend,
r#"
CREATE TABLE cpu (
hostname STRING PRIMARY KEY,
usage_user DOUBLE,
usage_system DOUBLE,
usage_idle DOUBLE,
greptime_timestamp TIMESTAMP TIME INDEX
) WITH (append_mode = 'true', sst_format = 'flat')
"#,
)
.await;
let mut rows = String::new();
for index in 0..1200 {
if index != 0 {
rows.push(',');
}
let hostname = match index % 3 {
0 => "host_a",
1 => "host_b",
_ => "host_c",
};
let usage_user = match index % 3 {
0 => 10.0,
1 => 20.0,
_ => 30.0,
};
let hour = 11 + index / 600;
let minute = (index % 600) / 60;
let second = index % 60;
rows.push_str(&format!(
"('{hostname}', {usage_user}, 1.0, 0.0, '2023-06-12T{hour:02}:{minute:02}:{second:02}Z')"
));
}
execute_sql(frontend, &format!("INSERT INTO cpu VALUES {rows}")).await;
execute_sql(frontend, "ADMIN FLUSH_TABLE('cpu')").await;
let count = execute_sql(frontend, "SELECT count(*) FROM cpu")
.await
.data
.pretty_print()
.await;
assert!(
count.contains("| 1200 |"),
"unexpected row count:\n{count}"
);
let query = r#"
SELECT hostname, avg(usage_user), date_trunc('hour', greptime_timestamp)
FROM cpu
WHERE greptime_timestamp >= '2023-06-12T00:00:00Z'
AND greptime_timestamp < '2023-06-14T00:00:00Z'
GROUP BY date_trunc('hour', greptime_timestamp), hostname
ORDER BY hostname, date_trunc('hour', greptime_timestamp)
"#;
let groups = execute_sql(frontend, query).await.data.pretty_print().await;
assert_eq!(
6,
groups
.lines()
.filter(|line| line.starts_with("| host_"))
.count(),
"expected one group per hostname and hour:\n{groups}"
);
for (hostname, average) in [("host_a", "10.0"), ("host_b", "20.0"), ("host_c", "30.0")] {
assert_eq!(
2,
groups.matches(&format!("| {hostname} ")).count(),
"expected two hourly groups for {hostname}:\n{groups}"
);
assert!(
groups.contains(average),
"expected average {average} for {hostname}:\n{groups}"
);
}
let explain = execute_sql(
frontend,
r#"
EXPLAIN ANALYZE VERBOSE
SELECT avg(usage_user), date_trunc('hour', greptime_timestamp)
FROM cpu
WHERE greptime_timestamp >= '2023-06-12T00:00:00Z'
AND greptime_timestamp < '2023-06-14T00:00:00Z'
GROUP BY date_trunc('hour', greptime_timestamp), hostname
"#,
)
.await
.data
.pretty_print()
.await;
assert!(
explain.contains("AggregateExec"),
"expected aggregate in physical plan:\n{explain}"
);
assert!(
!explain.contains("CastExec") && !explain.contains("CAST(hostname"),
"dictionary hostname must not be cast before aggregation:\n{explain}"
);
}
+9
View File
@@ -8004,6 +8004,15 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) {
),
"unexpected partial success body: {body:?}"
);
// The rejection must name the column, the failing value and the target type,
// otherwise locating the bad attribute needs a collector-side capture.
assert!(
partial_success.error_message.contains(
"failed to coerce trace column 'span_attributes.attr_int' in table \
'trace_type_abort' from String(\"not_a_number\") to Int64"
),
"unexpected partial success body: {body:?}"
);
validate_data(
"otlp_traces_v1_type_abort_rows",
+1
View File
@@ -16,6 +16,7 @@
mod admin_function_event;
mod database_ddl_event;
mod dict_groupby_sst;
mod event_recorder_test_util;
#[macro_use]
mod grpc;
@@ -0,0 +1,8 @@
name = "json2_storage_layout_settings"
reason = "Verify JSON2 storage layout settings in Arrow extension metadata survive restarts and upgrades."
introduced_by = "PR #8895"
topologies = ["distributed", "standalone"]
from_range = [">=v1.3.0"]
to_range = [">=v1.3.0"]
features = ["json2", "table", "metadata", "read"]
owner = "query"
@@ -0,0 +1,11 @@
CREATE TABLE t_json2_storage_layout_settings (
ts TIMESTAMP TIME INDEX,
j JSON2(max_auto_expanded_paths = 0)
) WITH (
'append_mode' = 'true'
);
INSERT INTO t_json2_storage_layout_settings VALUES
('2026-08-17 00:00:00+0000', '{"a": 1}');
ADMIN FLUSH_TABLE('t_json2_storage_layout_settings');
@@ -0,0 +1,27 @@
SELECT j.a AS a
FROM t_json2_storage_layout_settings;
+---+
| a |
+---+
| 1 |
+---+
SHOW CREATE TABLE t_json2_storage_layout_settings;
+---------------------------------+----------------------------------------------------------------+
| Table | Create Table |
+---------------------------------+----------------------------------------------------------------+
| t_json2_storage_layout_settings | CREATE TABLE IF NOT EXISTS "t_json2_storage_layout_settings" ( |
| | "ts" TIMESTAMP(3) NOT NULL, |
| | "j" JSON2( |
| | max_auto_expanded_paths = 0 |
| | ) NULL, |
| | TIME INDEX ("ts") |
| | ) |
| | |
| | ENGINE=mito |
| | WITH( |
| | append_mode = 'true' |
| | ) |
+---------------------------------+----------------------------------------------------------------+
@@ -0,0 +1,4 @@
SELECT j.a AS a
FROM t_json2_storage_layout_settings;
SHOW CREATE TABLE t_json2_storage_layout_settings;