From f3eb8e6c72e2c1583d5d4483ddc149c134966234 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Wed, 23 Sep 2026 01:52:52 +0000 Subject: [PATCH 1/2] test: cut integration test time and make the storage matrix meaningful (#9308) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test: cut integration test time and make the storage matrix meaningful tests-integration is ~85% of workspace test CPU, and 81% of that is the S3/S3WithCache variants of the HTTP and gRPC suites. Those suites do not touch the object store: of the 70 matrix HTTP tests only one flushed and read back an SST, so the matrix was paying real AWS round trips to re-prove protocol parsing. - Point the PR CI object-store matrix at the MinIO already started by tests-integration/fixtures. Three GT_S3_* consumers did not read GT_S3_ENDPOINT_URL and would have hit real AWS with MinIO credentials; they now do. - Add a nightly Linux job against real AWS S3, and pass GT_S3_* into the release integration-test container. The release previously ran every remote-backend case as a skip and only exercised the file backend. - Give each S3WithCache test its own read cache directory. They shared /tmp/greptimedb_cache, which the datanode wipes on startup, so a starting test deleted the read cache of a running one. - Add flush -> read-back assertions to the tests whose columns have a non-trivial SST representation: JSON/JSON2 columns, native histograms, metric-engine logical tables, and tables carrying fulltext or skipping indexes whose puffin files only exist after a flush. - Move eight tests that create no table out of the storage matrix. - Make the event recorder flush interval a constructor parameter and shorten it in the event tests, which otherwise wait a 5s window per DDL they assert on. It is skipped by serde and never reaches config files. - Drop duplicates: test_grpc_zstd_compression was a verbatim copy of test_grpc_message_size_ok and is now rewritten to assert the negotiated grpc-encoding; test_execute_copy_to_{s3,oss,gcs,azblob} were strict prefixes of their copy_from siblings; two standalone/distributed event test pairs shared one assertion body. - Fix and un-ignore stddev_by_label. stddev_pop merges partial aggregates in a parallelism-dependent order, so its last digits are unstable; the test now compares values with a tolerance. - Rebase the jaeger v1 fixture on the current instant. It carries ttl=7d with 2025 timestamps, so its rows were only readable as long as they stayed in the memtable. Signed-off-by: Dennis Zhuang * test: address review — wire nightly real-S3 job into check-status, keep the short event interval The nightly `check-status` job did not depend on the new real-S3 job, so a failure there would not have reached the status or Slack notification. In database_ddl_event the short interval was set by a first `with_event_recorder_options` call and then overwritten by the pre-existing one, which carries `..Default::default()`. Merged into a single call. Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang --- .../actions/build-linux-artifacts/action.yml | 23 + .github/workflows/dev-build.yml | 8 + .github/workflows/nightly-build.yml | 8 + .github/workflows/nightly-ci.yml | 53 ++- .github/workflows/release.yml | 8 + .github/workflows/rust.yml | 34 +- Makefile | 1 + src/common/event-recorder/src/recorder.rs | 30 +- src/frontend/src/instance/builder.rs | 1 + src/meta-srv/src/metasrv/builder.rs | 1 + src/object-store/src/test_util.rs | 29 -- src/object-store/tests/object_store_test.rs | 10 +- tests-integration/src/cluster.rs | 14 + tests-integration/src/test_util.rs | 79 ++-- tests-integration/src/tests/instance_test.rs | 198 +-------- tests-integration/src/tests/promql_test.rs | 49 ++- .../tests/admin_function_event.rs | 25 +- tests-integration/tests/database_ddl_event.rs | 7 +- .../tests/export_logical_tables.rs | 9 +- tests-integration/tests/flow_ddl_event.rs | 5 +- tests-integration/tests/gc_event.rs | 3 + tests-integration/tests/grpc.rs | 147 ++++++- tests-integration/tests/http.rs | 396 ++++++++++++++++-- tests-integration/tests/main.rs | 11 + .../tests/reconciliation_event.rs | 4 + tests-integration/tests/repartition_event.rs | 2 + tests-integration/tests/table_ddl_event.rs | 5 +- tests-integration/tests/view_ddl_event.rs | 16 +- tests-integration/tests/wal_prune_event.rs | 2 + 29 files changed, 805 insertions(+), 373 deletions(-) diff --git a/.github/actions/build-linux-artifacts/action.yml b/.github/actions/build-linux-artifacts/action.yml index 3cb9c439551..c5212c28998 100644 --- a/.github/actions/build-linux-artifacts/action.yml +++ b/.github/actions/build-linux-artifacts/action.yml @@ -31,6 +31,22 @@ inputs: description: Build GreptimeDB with large page size (65536). required: false default: 'false' + aws-ci-test-bucket: + description: Bucket used by the object-store integration tests + required: false + default: '' + aws-region: + description: Region of the test bucket + required: false + default: '' + aws-access-key-id: + description: Access key id for the test bucket + required: false + default: '' + aws-secret-access-key: + description: Secret access key for the test bucket + required: false + default: '' runs: using: composite steps: @@ -38,6 +54,13 @@ runs: if: ${{ inputs.disable-run-tests == 'false' }} shell: bash # NOTE: If the BUILD_JOBS > 4, it's always OOM in EC2 instance. + # Without GT_S3_*, every remote object-store case is skipped and the + # release only ever exercises the local file backend. + env: + GT_S3_BUCKET: ${{ inputs.aws-ci-test-bucket }} + GT_S3_REGION: ${{ inputs.aws-region }} + GT_S3_ACCESS_KEY_ID: ${{ inputs.aws-access-key-id }} + GT_S3_ACCESS_KEY: ${{ inputs.aws-secret-access-key }} run: | cd ${{ inputs.working-dir }} && \ make run-it-in-container BUILD_JOBS=4 \ diff --git a/.github/workflows/dev-build.yml b/.github/workflows/dev-build.yml index d03fbeff14a..563435b708e 100644 --- a/.github/workflows/dev-build.yml +++ b/.github/workflows/dev-build.yml @@ -183,6 +183,10 @@ jobs: image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} large-page-size: ${{ inputs.large-page-size }} + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} build-linux-arm64-artifacts: name: Build linux-arm64 artifacts @@ -217,6 +221,10 @@ jobs: image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} large-page-size: ${{ inputs.large-page-size }} + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} release-images-to-dockerhub: name: Build and push images to DockerHub diff --git a/.github/workflows/nightly-build.yml b/.github/workflows/nightly-build.yml index 14ebb6e715e..c78ca4dc5b9 100644 --- a/.github/workflows/nightly-build.yml +++ b/.github/workflows/nightly-build.yml @@ -151,6 +151,10 @@ jobs: disable-run-tests: ${{ env.DISABLE_RUN_TESTS }} image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} build-linux-arm64-artifacts: name: Build linux-arm64 artifacts @@ -173,6 +177,10 @@ jobs: disable-run-tests: ${{ env.DISABLE_RUN_TESTS }} image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} release-images-to-dockerhub: name: Build and push images to DockerHub diff --git a/.github/workflows/nightly-ci.yml b/.github/workflows/nightly-ci.yml index d4afa2778b8..8441ed02312 100644 --- a/.github/workflows/nightly-ci.yml +++ b/.github/workflows/nightly-ci.yml @@ -71,6 +71,57 @@ jobs: path: C:\Users\RUNNER~1\AppData\Local\Temp\sqlness* retention-days: 3 + # PR CI runs the object-store matrix against a local MinIO. This job is where + # the same matrix meets real AWS S3. + test-on-linux-real-s3: + name: Run tests on Linux against real S3 + if: ${{ github.repository == 'GreptimeTeam/greptimedb' }} + runs-on: ubuntu-22.04-arm + timeout-minutes: 90 + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: arduino/setup-protoc@v3 + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} + - uses: rui314/setup-mold@v1 + - name: Install toolchain + uses: actions-rust-lang/setup-rust-toolchain@v1 + with: + cache: false + - name: Rust Cache + uses: Swatinem/rust-cache@v2 + with: + shared-key: "coverage-test" + cache-all-crates: "true" + save-if: false + - name: Install latest nextest release + uses: taiki-e/install-action@nextest + - name: Setup external services + working-directory: tests-integration/fixtures + run: ../../.github/scripts/pull-test-deps-images.sh && docker compose up -d --wait + - name: Run nextest cases + run: cargo nextest run --workspace -F dashboard -F pg_kvbackend -F mysql_kvbackend -F vector_index -F ai_functions + env: + CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold" + RUST_BACKTRACE: 1 + RUST_MIN_STACK: 8388608 # 8MB + CARGO_INCREMENTAL: 0 + GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }} + GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} + GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + GT_ETCD_TLS_ENDPOINTS: https://127.0.0.1:2378 + GT_ETCD_ENDPOINTS: http://127.0.0.1:2379 + GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres + GT_POSTGRES15_ENDPOINTS: postgres://test_user:test_password@127.0.0.1:5433/postgres + GT_POSTGRES15_SCHEMA: test_schema + GT_MYSQL_ENDPOINTS: mysql://greptimedb:admin@127.0.0.1:3306/mysql + GT_KAFKA_ENDPOINTS: 127.0.0.1:9092 + GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093 + UNITTEST_LOG_DIR: "__unittest_logs" + test-on-windows: name: Run tests on Windows if: ${{ github.repository == 'GreptimeTeam/greptimedb' }} @@ -130,7 +181,7 @@ jobs: check-status: name: Check status - needs: [sqlness-test, sqlness-windows, test-on-windows] + needs: [sqlness-test, sqlness-windows, test-on-windows, test-on-linux-real-s3] if: ${{ github.repository == 'GreptimeTeam/greptimedb' }} runs-on: ubuntu-latest outputs: diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index acf52e241c2..6c0c36c895e 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -204,6 +204,10 @@ jobs: disable-run-tests: ${{ env.DISABLE_RUN_TESTS }} image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} build-linux-arm64-artifacts: name: Build linux-arm64 artifacts @@ -227,6 +231,10 @@ jobs: image-registry: ${{ vars.ECR_IMAGE_REGISTRY }} image-namespace: ${{ vars.ECR_IMAGE_NAMESPACE }} large-page-size: true + aws-ci-test-bucket: ${{ vars.AWS_CI_TEST_BUCKET }} + aws-region: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} + aws-access-key-id: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} build-linux-riscv64-artifacts: name: Build linux-riscv64 artifacts diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 1a093a2d2a5..e4483b4a42b 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -206,15 +206,14 @@ jobs: RUST_BACKTRACE: 1 RUST_MIN_STACK: 8388608 # 8MB CARGO_INCREMENTAL: 0 - GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }} - GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} - GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} - GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} - GT_MINIO_BUCKET: greptime - GT_MINIO_ACCESS_KEY_ID: superpower_ci_user - GT_MINIO_ACCESS_KEY: superpower_password - GT_MINIO_REGION: us-west-2 - GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000 + # Runs the object-store matrix against the local MinIO from + # tests-integration/fixtures. Real AWS S3 is covered by the nightly + # jobs and by the release build. + GT_S3_BUCKET: greptime + GT_S3_ACCESS_KEY_ID: superpower_ci_user + GT_S3_ACCESS_KEY: superpower_password + GT_S3_REGION: us-west-2 + GT_S3_ENDPOINT_URL: http://127.0.0.1:9000 GT_ETCD_TLS_ENDPOINTS: https://127.0.0.1:2378 GT_ETCD_ENDPOINTS: http://127.0.0.1:2379 GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres @@ -263,15 +262,14 @@ jobs: CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold" RUST_BACKTRACE: 1 CARGO_INCREMENTAL: 0 - GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }} - GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }} - GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }} - GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }} - GT_MINIO_BUCKET: greptime - GT_MINIO_ACCESS_KEY_ID: superpower_ci_user - GT_MINIO_ACCESS_KEY: superpower_password - GT_MINIO_REGION: us-west-2 - GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000 + # Runs the object-store matrix against the local MinIO from + # tests-integration/fixtures. Real AWS S3 is covered by the nightly + # jobs and by the release build. + GT_S3_BUCKET: greptime + GT_S3_ACCESS_KEY_ID: superpower_ci_user + GT_S3_ACCESS_KEY: superpower_password + GT_S3_REGION: us-west-2 + GT_S3_ENDPOINT_URL: http://127.0.0.1:9000 GT_ETCD_TLS_ENDPOINTS: https://127.0.0.1:2378 GT_ETCD_ENDPOINTS: http://127.0.0.1:2379 GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres diff --git a/Makefile b/Makefile index 29801465351..7176eafd0bb 100644 --- a/Makefile +++ b/Makefile @@ -264,6 +264,7 @@ stop-etcd: ## Stop single node etcd for testing purpose. run-it-in-container: start-etcd ## Run integration tests in dev-builder. docker run --network=host \ -v ${PWD}:/greptimedb -v ${CARGO_REGISTRY_CACHE}:/root/.cargo/registry -v ${CARGO_GIT_CACHE}:/root/.cargo/git -v /tmp:/tmp \ + -e GT_S3_BUCKET -e GT_S3_ACCESS_KEY_ID -e GT_S3_ACCESS_KEY -e GT_S3_REGION \ -w /greptimedb ${IMAGE_REGISTRY}/${IMAGE_NAMESPACE}/dev-builder-${BASE_IMAGE}:${DEV_BUILDER_IMAGE_TAG} \ make test sqlness-test BUILD_JOBS=${BUILD_JOBS} diff --git a/src/common/event-recorder/src/recorder.rs b/src/common/event-recorder/src/recorder.rs index b6c18ed08de..3ea5c7e55a8 100644 --- a/src/common/event-recorder/src/recorder.rs +++ b/src/common/event-recorder/src/recorder.rs @@ -291,17 +291,29 @@ pub struct EventRecorderOptions { skip_serializing_if = "event_type_filter_is_all" )] pub event_types: EventTypeFilterRef, + /// How long buffered events may wait before reaching the event handler. + /// + /// Not a configuration option: it is skipped by serde so it never appears in + /// config files or `config/config.md`. Integration tests shorten it so they + /// do not pay a flush window per DDL they assert on. + #[serde(skip, default = "default_flush_interval")] + pub flush_interval: Duration, } fn default_events_table_ttl() -> Duration { DEFAULT_EVENTS_TABLE_TTL } +fn default_flush_interval() -> Duration { + DEFAULT_FLUSH_INTERVAL_SECONDS +} + impl Default for EventRecorderOptions { fn default() -> Self { Self { ttl: DEFAULT_EVENTS_TABLE_TTL, event_types: Arc::new(EventTypeFilter::All), + flush_interval: DEFAULT_FLUSH_INTERVAL_SECONDS, } } } @@ -321,13 +333,18 @@ pub struct EventRecorderImpl { impl EventRecorderImpl { pub fn new(event_handler: Box) -> Self { - Self::with_event_type_filter(event_handler, Arc::new(EventTypeFilter::All)) + Self::with_event_type_filter( + event_handler, + Arc::new(EventTypeFilter::All), + DEFAULT_FLUSH_INTERVAL_SECONDS, + ) } /// Creates an event recorder with an event-type filter. pub fn with_event_type_filter( event_handler: Box, event_types: EventTypeFilterRef, + flush_interval: Duration, ) -> Self { let (tx, rx) = channel(DEFAULT_CHANNEL_SIZE); let cancel_token = CancellationToken::new(); @@ -339,13 +356,9 @@ impl EventRecorderImpl { cancel_token: cancel_token.clone(), }; - let processor = EventProcessor::new( - rx, - event_handler, - DEFAULT_FLUSH_INTERVAL_SECONDS, - DEFAULT_MAX_RETRY_TIMES, - ) - .with_cancel_token(cancel_token); + let processor = + EventProcessor::new(rx, event_handler, flush_interval, DEFAULT_MAX_RETRY_TIMES) + .with_cancel_token(cancel_token); // Spawn a background task to process the events. let handle = tokio::spawn(async move { @@ -611,6 +624,7 @@ mod tests { count: count.clone(), }), event_type_filter.clone(), + DEFAULT_FLUSH_INTERVAL_SECONDS, ); assert!(Arc::ptr_eq( diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index a3cfbcdaffe..d9b86ac5a05 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -381,6 +381,7 @@ impl FrontendBuilder { self.options.event_recorder.ttl, )), self.options.event_recorder.event_types.clone(), + self.options.event_recorder.flush_interval, )); admin_event_recorder.install(&event_recorder); diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs index b40f06e3cfe..cbc859716a4 100644 --- a/src/meta-srv/src/metasrv/builder.rs +++ b/src/meta-srv/src/metasrv/builder.rs @@ -250,6 +250,7 @@ impl MetasrvBuilder { let event_recorder = Arc::new(EventRecorderImpl::with_event_type_filter( Box::new(EventHandlerImpl::new(event_inserter)), options.event_recorder.event_types.clone(), + options.event_recorder.flush_interval, )); let selector = selector.unwrap_or_else(|| Arc::new(LeaseBasedSelector)); diff --git a/src/object-store/src/test_util.rs b/src/object-store/src/test_util.rs index af68978b0e4..936ecba5bac 100644 --- a/src/object-store/src/test_util.rs +++ b/src/object-store/src/test_util.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::env; - use crate::{ObjectStore, Result}; /// Temp folder for object store test @@ -35,30 +33,3 @@ impl TempFolder { self.store.delete_with(&self.path).recursive(true).await } } - -/// Test s3 config from environment variables -#[derive(Debug)] -pub struct TestS3Config { - pub root: String, - pub access_key_id: String, - pub secret_access_key: String, - pub bucket: String, - pub region: Option, -} - -/// Returns s3 test config, return None if not found. -pub fn s3_test_config() -> Option { - if let Ok(b) = env::var("GT_S3_BUCKET") - && !b.is_empty() - { - return Some(TestS3Config { - root: uuid::Uuid::new_v4().to_string(), - access_key_id: env::var("GT_S3_ACCESS_KEY_ID").ok()?, - secret_access_key: env::var("GT_S3_ACCESS_KEY").ok()?, - bucket: env::var("GT_S3_BUCKET").ok()?, - region: Some(env::var("GT_S3_REGION").ok()?), - }); - } - - None -} diff --git a/src/object-store/tests/object_store_test.rs b/src/object-store/tests/object_store_test.rs index 9cf03fe42c5..735d2c56f15 100644 --- a/src/object-store/tests/object_store_test.rs +++ b/src/object-store/tests/object_store_test.rs @@ -284,13 +284,21 @@ async fn test_s3_backend() -> Result<()> { let root = uuid::Uuid::new_v4().to_string(); - let builder = S3::default() + let mut builder = S3::default() .root(&root) .access_key_id(&env::var("GT_S3_ACCESS_KEY_ID")?) .secret_access_key(&env::var("GT_S3_ACCESS_KEY")?) .region(&env::var("GT_S3_REGION")?) .bucket(&bucket); + // Honors an S3-compatible endpoint (MinIO in CI) so the test does not + // fall back to resolving the bucket against real AWS. + if let Ok(endpoint) = env::var("GT_S3_ENDPOINT_URL") + && !endpoint.is_empty() + { + builder = builder.endpoint(&endpoint); + } + let store = ObjectStore::new(builder).unwrap(); let store = object_store::util::with_instrument_layers(store, false); diff --git a/tests-integration/src/cluster.rs b/tests-integration/src/cluster.rs index 3e53b6104a0..bc2ed6f20a4 100644 --- a/tests-integration/src/cluster.rs +++ b/tests-integration/src/cluster.rs @@ -36,6 +36,7 @@ use client::client_manager::NodeClients; use cmd::frontend::create_heartbeat_task; use common_base::Plugins; use common_datasource::object_store::LocalFileAccess; +use common_event_recorder::EventRecorderOptions; use common_grpc::channel_manager::{ChannelConfig, ChannelManager}; use common_meta::DatanodeId; use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder}; @@ -174,6 +175,7 @@ pub struct GreptimeDbClusterBuilder { shared_home_dir: Option>, meta_selector: Option, local_file_access: LocalFileAccess, + event_recorder_options: EventRecorderOptions, } impl GreptimeDbClusterBuilder { @@ -209,6 +211,7 @@ impl GreptimeDbClusterBuilder { shared_home_dir: None, meta_selector: None, local_file_access: LocalFileAccess::default(), + event_recorder_options: EventRecorderOptions::default(), } } @@ -253,6 +256,15 @@ impl GreptimeDbClusterBuilder { self } + #[must_use] + pub fn with_event_recorder_options( + mut self, + event_recorder_options: EventRecorderOptions, + ) -> Self { + self.event_recorder_options = event_recorder_options; + self + } + /// Sets whether the test frontend automatically creates tables on write. #[must_use] pub fn with_frontend_auto_create_table(mut self, auto_create_table: bool) -> Self { @@ -303,6 +315,7 @@ impl GreptimeDbClusterBuilder { ..Default::default() }, gc: self.metasrv_gc_config.clone(), + event_recorder: self.event_recorder_options.clone(), ..Default::default() }; @@ -550,6 +563,7 @@ impl GreptimeDbClusterBuilder { fn build_frontend_options(&self) -> FrontendOptions { let mut fe_opts = FrontendOptions { auto_create_table: self.frontend_auto_create_table, + event_recorder: self.event_recorder_options.clone(), ..Default::default() }; diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index 562bd48cdb0..fd02db2e5a1 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -31,6 +31,7 @@ use client::{Client, Database}; use common_base::Plugins; use common_catalog::consts::MIN_USER_TABLE_ID; use common_config::Configurable; +use common_event_recorder::EventRecorderOptions; #[cfg(test)] use common_meta::DatanodeId; use common_meta::key::TableMetadataManager; @@ -328,6 +329,18 @@ impl StorageType { } } +/// Event recorder options for tests. +/// +/// The production flush interval is 5s, and the event tests interleave "run a +/// DDL, wait for its event, run the next DDL", so each barrier costs a full +/// window. A short interval removes that wait without changing what is asserted. +pub fn test_event_recorder_options() -> EventRecorderOptions { + EventRecorderOptions { + flush_interval: Duration::from_millis(100), + ..Default::default() + } +} + fn s3_test_config() -> S3Config { S3Config { connection: S3Connection { @@ -363,7 +376,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Gcs::from(&gcs_config.connection); let config = ObjectStoreConfig::Gcs(gcs_config); let store = ObjectStore::new(builder).unwrap(); - (config, TempDirGuard::Gcs(TempFolder::new(&store, "/"))) + (config, TempDirGuard::remote(TempFolder::new(&store, "/"))) } StorageType::Azblob => { let azblob_config = AzblobConfig { @@ -381,7 +394,7 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Azblob::from(&azblob_config.connection); let config = ObjectStoreConfig::Azblob(azblob_config); let store = ObjectStore::new(builder).unwrap(); - (config, TempDirGuard::Azblob(TempFolder::new(&store, "/"))) + (config, TempDirGuard::remote(TempFolder::new(&store, "/"))) } StorageType::Oss => { let oss_config = OssConfig { @@ -398,32 +411,57 @@ pub fn get_test_store_config(store_type: &StorageType) -> (ObjectStoreConfig, Te let builder = Oss::from(&oss_config.connection); let config = ObjectStoreConfig::Oss(oss_config); let store = ObjectStore::new(builder).unwrap(); - (config, TempDirGuard::Oss(TempFolder::new(&store, "/"))) + (config, TempDirGuard::remote(TempFolder::new(&store, "/"))) } StorageType::S3 | StorageType::S3WithCache => { let mut s3_config = s3_test_config(); - if *store_type == StorageType::S3WithCache { - s3_config.cache.cache_path = "/tmp/greptimedb_cache".to_string(); + // The datanode wipes `/cache/object/read` on startup, so a + // path shared between concurrently running tests lets a starting test + // delete the read cache of a running one. + let cache_dir = if *store_type == StorageType::S3WithCache { + let dir = create_temp_dir("gt_s3_read_cache"); + s3_config.cache.cache_path = dir.path().to_string_lossy().to_string(); + Some(dir) } else { s3_config.cache.enable_read_cache = false; - } + None + }; let builder = S3::from(&s3_config.connection); let config = ObjectStoreConfig::S3(s3_config); let store = ObjectStore::new(builder).unwrap(); - (config, TempDirGuard::S3(TempFolder::new(&store, "/"))) + ( + config, + TempDirGuard { + remote: Some(TempFolder::new(&store, "/")), + local_cache: cache_dir, + }, + ) } - StorageType::File => (ObjectStoreConfig::File(FileConfig {}), TempDirGuard::None), + StorageType::File => ( + ObjectStoreConfig::File(FileConfig {}), + TempDirGuard::default(), + ), } } -pub enum TempDirGuard { - None, - S3(TempFolder), - Oss(TempFolder), - Azblob(TempFolder), - Gcs(TempFolder), +#[derive(Default)] +pub struct TempDirGuard { + /// Prefix to wipe from the remote object store, absent for the file backend. + remote: Option, + /// Local read cache directory. Only held so it is removed when the guard drops. + #[allow(dead_code)] + local_cache: Option, +} + +impl TempDirGuard { + fn remote(folder: TempFolder) -> Self { + Self { + remote: Some(folder), + local_cache: None, + } + } } pub struct TestGuard { @@ -446,11 +484,7 @@ pub struct StorageGuard(pub TempDirGuard); impl TestGuard { pub async fn remove_all(&mut self) { for storage_guard in self.storage_guards.iter_mut() { - if let TempDirGuard::S3(guard) - | TempDirGuard::Oss(guard) - | TempDirGuard::Azblob(guard) - | TempDirGuard::Gcs(guard) = &mut storage_guard.0 - { + if let Some(guard) = &mut storage_guard.0.remote { guard.remove_all().await.unwrap() } } @@ -465,11 +499,8 @@ impl Drop for TestGuard { common_runtime::spawn_global(async move { let mut errors = vec![]; for guard in guards { - if let TempDirGuard::S3(guard) - | TempDirGuard::Oss(guard) - | TempDirGuard::Azblob(guard) - | TempDirGuard::Gcs(guard) = guard.0 - && let Err(e) = guard.remove_all().await + if let Some(remote) = guard.0.remote + && let Err(e) = remote.remove_all().await { errors.push(e); } diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index dc4d3898860..caac6d33238 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -2195,47 +2195,6 @@ async fn test_delete(instance: Arc) { check_output_stream(output, expect).await; } -#[apply(both_instances_cases)] -async fn test_execute_copy_to_s3(instance: Arc) { - if let Ok(bucket) = env::var("GT_S3_BUCKET") - && !bucket.is_empty() - { - let instance = instance.frontend(); - - // setups - assert!(matches!(execute_sql( - &instance, - "create table demo(host string, cpu double, memory double, ts timestamp time index);", - ) - .await.data, OutputData::AffectedRows(0))); - - let output = execute_sql( - &instance, - r#"insert into demo(host, cpu, memory, ts) values - ('host1', 66.6, 1024, 1655276557000), - ('host2', 88.8, 333.3, 1655276558000) - "#, - ) - .await - .data; - assert!(matches!(output, OutputData::AffectedRows(2))); - let key_id = env::var("GT_S3_ACCESS_KEY_ID").unwrap(); - let key = env::var("GT_S3_ACCESS_KEY").unwrap(); - let region = env::var("GT_S3_REGION").unwrap(); - - let root = uuid::Uuid::new_v4().to_string(); - - // exports - let copy_to_stmt = format!( - "Copy demo TO 's3://{}/{}/export/demo.parquet' CONNECTION (ACCESS_KEY_ID='{}',SECRET_ACCESS_KEY='{}',REGION='{}')", - bucket, root, key_id, key, region - ); - - let output = execute_sql(&instance, ©_to_stmt).await.data; - assert!(matches!(output, OutputData::AffectedRows(2))); - } -} - #[apply(both_instances_cases)] async fn test_execute_copy_from_s3(instance: Arc) { common_telemetry::init_default_ut_logging(); @@ -2268,9 +2227,21 @@ async fn test_execute_copy_from_s3(instance: Arc) { let key = env::var("GT_S3_ACCESS_KEY").unwrap(); let region = env::var("GT_S3_REGION").unwrap(); + let mut connection = format!( + "ACCESS_KEY_ID='{}',SECRET_ACCESS_KEY='{}',REGION='{}'", + key_id, key, region + ); + // Honors an S3-compatible endpoint (MinIO in CI) so the test does not + // fall back to resolving the bucket against real AWS. + if let Ok(endpoint) = env::var("GT_S3_ENDPOINT_URL") + && !endpoint.is_empty() + { + connection = format!("{},ENDPOINT='{}'", connection, endpoint); + } + let copy_to_stmt = format!( - "Copy demo TO 's3://{}/{}/export/demo.parquet' CONNECTION (ACCESS_KEY_ID='{}',SECRET_ACCESS_KEY='{}',REGION='{}')", - bucket, root, key_id, key, region + "Copy demo TO 's3://{}/{}/export/demo.parquet' CONNECTION ({})", + bucket, root, connection ); let output = execute_sql(&instance, ©_to_stmt).await.data; @@ -2315,10 +2286,7 @@ async fn test_execute_copy_from_s3(instance: Arc) { .data, OutputData::AffectedRows(0) )); - let sql = format!( - "{} CONNECTION (ACCESS_KEY_ID='{}',SECRET_ACCESS_KEY='{}',REGION='{}')", - test.sql, key_id, key, region, - ); + let sql = format!("{} CONNECTION ({})", test.sql, connection); let output = execute_sql(&instance, &sql).await.data; assert!(matches!(output, OutputData::AffectedRows(2))); @@ -2340,50 +2308,6 @@ async fn test_execute_copy_from_s3(instance: Arc) { } } -#[apply(both_instances_cases)] -async fn test_execute_copy_to_oss(instance: Arc) { - if let Ok(bucket) = env::var("GT_OSS_BUCKET") - && !bucket.is_empty() - { - let instance = instance.frontend(); - - assert!(matches!( - execute_sql( - &instance, - "create table demo(host string, cpu double, memory double, ts timestamp time index);", - ) - .await - .data, - OutputData::AffectedRows(0) - )); - - let output = execute_sql( - &instance, - r#"insert into demo(host, cpu, memory, ts) values - ('host1', 66.6, 1024, 1655276557000), - ('host2', 88.8, 333.3, 1655276558000) - "#, - ) - .await - .data; - assert!(matches!(output, OutputData::AffectedRows(2))); - - let key_id = env::var("GT_OSS_ACCESS_KEY_ID").unwrap(); - let key = env::var("GT_OSS_ACCESS_KEY").unwrap(); - let endpoint = env::var("GT_OSS_ENDPOINT").unwrap(); - - let root = uuid::Uuid::new_v4().to_string(); - - let copy_to_stmt = format!( - "Copy demo TO 'oss://{}/{}/export/demo.parquet' CONNECTION (ACCESS_KEY_ID='{}',ACCESS_KEY_SECRET='{}',ENDPOINT='{}')", - bucket, root, key_id, key, endpoint - ); - - let output = execute_sql(&instance, ©_to_stmt).await.data; - assert!(matches!(output, OutputData::AffectedRows(2))); - } -} - #[apply(both_instances_cases)] async fn test_execute_copy_from_oss(instance: Arc) { common_telemetry::init_default_ut_logging(); @@ -2490,54 +2414,6 @@ async fn test_execute_copy_from_oss(instance: Arc) { } } -#[apply(both_instances_cases)] -async fn test_execute_copy_to_gcs(instance: Arc) { - if let (Ok(bucket), Ok(scope), Ok(credential)) = ( - env::var("GT_GCS_BUCKET"), - env::var("GT_GCS_SCOPE"), - env::var("GT_GCS_CREDENTIAL"), - ) && !bucket.is_empty() - { - let endpoint = env::var("GT_GCS_ENDPOINT").unwrap_or_default(); - let instance = instance.frontend(); - - assert!(matches!( - execute_sql( - &instance, - "create table demo(host string, cpu double, memory double, ts timestamp time index);", - ) - .await - .data, - OutputData::AffectedRows(0) - )); - - let output = execute_sql( - &instance, - r#"insert into demo(host, cpu, memory, ts) values - ('host1', 66.6, 1024, 1655276557000), - ('host2', 88.8, 333.3, 1655276558000) - "#, - ) - .await - .data; - assert!(matches!(output, OutputData::AffectedRows(2))); - - let root = uuid::Uuid::new_v4().to_string(); - let mut connection = format!("SCOPE='{}',CREDENTIAL='{}'", scope, credential); - if !endpoint.is_empty() { - connection = format!("{},ENDPOINT='{}'", connection, endpoint); - } - - let copy_to_stmt = format!( - "Copy demo TO 'gcs://{}/{}/export/demo.parquet' CONNECTION ({})", - bucket, root, connection - ); - - let output = execute_sql(&instance, ©_to_stmt).await.data; - assert!(matches!(output, OutputData::AffectedRows(2))); - } -} - #[apply(both_instances_cases)] async fn test_execute_copy_from_gcs(instance: Arc) { common_telemetry::init_default_ut_logging(); @@ -2646,50 +2522,6 @@ async fn test_execute_copy_from_gcs(instance: Arc) { } } -#[apply(both_instances_cases)] -async fn test_execute_copy_to_azblob(instance: Arc) { - if let (Ok(container), Ok(account_name), Ok(account_key), Ok(endpoint)) = ( - env::var("GT_AZBLOB_CONTAINER"), - env::var("GT_AZBLOB_ACCOUNT_NAME"), - env::var("GT_AZBLOB_ACCOUNT_KEY"), - env::var("GT_AZBLOB_ENDPOINT"), - ) && !container.is_empty() - { - let instance = instance.frontend(); - - assert!(matches!( - execute_sql( - &instance, - "create table demo(host string, cpu double, memory double, ts timestamp time index);", - ) - .await - .data, - OutputData::AffectedRows(0) - )); - - let output = execute_sql( - &instance, - r#"insert into demo(host, cpu, memory, ts) values - ('host1', 66.6, 1024, 1655276557000), - ('host2', 88.8, 333.3, 1655276558000) - "#, - ) - .await - .data; - assert!(matches!(output, OutputData::AffectedRows(2))); - - let root = uuid::Uuid::new_v4().to_string(); - - let copy_to_stmt = format!( - "Copy demo TO 'azblob://{}/{}/export/demo.parquet' CONNECTION (ACCOUNT_NAME='{}',ACCOUNT_KEY='{}',ENDPOINT='{}')", - container, root, account_name, account_key, endpoint - ); - - let output = execute_sql(&instance, ©_to_stmt).await.data; - assert!(matches!(output, OutputData::AffectedRows(2))); - } -} - #[apply(both_instances_cases)] async fn test_execute_copy_from_azblob(instance: Arc) { common_telemetry::init_default_ut_logging(); diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index c816583f135..74114bbd6c4 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -644,28 +644,57 @@ async fn two_aggregators_combined_aggrs(instance: Arc) { // eval instant at 50m stddev by (instance)(http_requests) // {instance="0"} 223.60679774998 // {instance="1"} 223.60679774998 +// +// The values are compared with a tolerance rather than against a rendered table: +// `stddev_pop` merges partial aggregates in a parallelism-dependent order, so the +// last digits of its output are not stable across runs. #[apply(both_instances_cases)] -#[ignore = "TODO(ruihang): fix this case"] async fn stddev_by_label(instance: Arc) { let instance = instance.frontend(); - create_insert_query_assert( + execute_all(&instance, AGGREGATORS_CREATE_TABLE, QueryContext::arc()).await; + execute_all(&instance, AGGREGATORS_INSERT_DATA, QueryContext::arc()).await; + + let batches = promql_query_as_batches( instance, - AGGREGATORS_CREATE_TABLE, - AGGREGATORS_INSERT_DATA, r#"stddev by (instance)(http_requests)"#, + None, + QueryContext::arc(), UNIX_EPOCH, unix_epoch_plus_100s(), Duration::from_secs(60), Duration::from_secs(0), - "+----------+---------------------+--------------------------------+\ - \n| instance | ts | STDDEVPOP(http_requests.value) |\ - \n+----------+---------------------+--------------------------------+\ - \n| 0 | 1970-01-01T00:00:00 | 223.606797749979 |\ - \n| 1 | 1970-01-01T00:00:00 | 223.606797749979 |\ - \n+----------+---------------------+--------------------------------+", ) .await; + + let batches = batches.iter().collect::>(); + assert_eq!(batches.len(), 1); + // Tag columns come back dictionary-encoded. + let instances = cast( + batches[0].column_by_name("instance").unwrap(), + &DataType::Utf8, + ) + .unwrap(); + let instances = instances.as_any().downcast_ref::().unwrap(); + let values = batches[0] + .column_by_name("stddev_pop(http_requests.value)") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + + let mut actual = (0..values.len()) + .map(|i| (instances.value(i), values.value(i))) + .collect::>(); + actual.sort_by(|a, b| a.0.cmp(b.0)); + assert_eq!(actual.len(), 2, "{}", batches[0].pretty_print()); + for (expected_instance, (instance, value)) in ["0", "1"].into_iter().zip(actual) { + assert_eq!(instance, expected_instance); + assert!( + (value - 223.606_797_749_979).abs() < 1e-9, + "instance {instance}: {value}" + ); + } } // This is not derived from prometheus diff --git a/tests-integration/tests/admin_function_event.rs b/tests-integration/tests/admin_function_event.rs index d669d2900ad..43f7445ca2f 100644 --- a/tests-integration/tests/admin_function_event.rs +++ b/tests-integration/tests/admin_function_event.rs @@ -14,7 +14,6 @@ use std::sync::Arc; -use common_test_util::temp_dir::create_temp_dir; use frontend::instance::Instance; use servers::query_handler::sql::SqlQueryHandler; use session::context::QueryContext; @@ -23,9 +22,8 @@ use sql::dialect::GreptimeDbDialect; use sql::parser::{ParseOptions, ParserContext}; use sql::statements::admin::Admin; use sql::statements::statement::Statement; -use tests_integration::cluster::GreptimeDbClusterBuilder; use tests_integration::standalone::GreptimeDbStandaloneBuilder; -use tests_integration::test_util::{StorageType, get_test_store_config}; +use tests_integration::test_util::test_event_recorder_options; use crate::event_recorder_test_util::assert_eventually_eq; @@ -35,31 +33,12 @@ const TABLE_NAME: &str = "admin_function_event_test"; async fn test_standalone_admin_function_events() { common_telemetry::init_default_ut_logging(); let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_admin_function_events") + .with_event_recorder_options(test_event_recorder_options()) .build() .await; assert_admin_function_events(standalone.fe_instance()).await; } -#[tokio::test(flavor = "multi_thread")] -async fn test_distributed_admin_function_events() { - let store_type = StorageType::File; - if !store_type.test_on() { - return; - } - - common_telemetry::init_default_ut_logging(); - let (store_config, _guard) = get_test_store_config(&store_type); - let home_dir = create_temp_dir("test_distributed_admin_function_events_data_home"); - let cluster = GreptimeDbClusterBuilder::new("test_distributed_admin_function_events") - .await - .with_datanodes(1) - .with_store_config(store_config) - .with_shared_home_dir(Arc::new(home_dir)) - .build(true) - .await; - assert_admin_function_events(cluster.fe_instance()).await; -} - async fn assert_admin_function_events(instance: &Arc) { execute( instance, diff --git a/tests-integration/tests/database_ddl_event.rs b/tests-integration/tests/database_ddl_event.rs index c08b81211de..1c6d3f39633 100644 --- a/tests-integration/tests/database_ddl_event.rs +++ b/tests-integration/tests/database_ddl_event.rs @@ -21,6 +21,7 @@ use tests_integration::cluster::GreptimeDbClusterBuilder; use tests_integration::standalone::GreptimeDbStandaloneBuilder; use tests_integration::test_util::{ StorageType, execute_sql, get_test_store_config, setup_authenticated_grpc_database, + test_event_recorder_options, }; use crate::event_recorder_test_util::{ @@ -42,7 +43,7 @@ async fn test_event_table_auto_creation_with_auto_create_disabled() { event_types: Arc::new(EventTypeFilter::Only( [String::from("create_database")].into_iter().collect(), )), - ..Default::default() + ..test_event_recorder_options() }) .build() .await; @@ -94,7 +95,7 @@ async fn test_event_table_schema_reconciliation_with_auto_create_disabled() { event_types: Arc::new(EventTypeFilter::Only( [String::from("create_database")].into_iter().collect(), )), - ..Default::default() + ..test_event_recorder_options() }) .build() .await; @@ -185,6 +186,7 @@ async fn test_database_ddl_events() { let home_dir = create_temp_dir("test_database_ddl_events_data_home"); let cluster = GreptimeDbClusterBuilder::new("test_database_ddl_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .with_frontend_auto_create_table(false) .with_store_config(store_config) @@ -208,6 +210,7 @@ async fn test_database_ddl_events() { async fn test_standalone_database_ddl_events() { common_telemetry::init_default_ut_logging(); let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_database_ddl_events") + .with_event_recorder_options(test_event_recorder_options()) .build() .await; let instance = standalone.fe_instance(); diff --git a/tests-integration/tests/export_logical_tables.rs b/tests-integration/tests/export_logical_tables.rs index 92888cced3c..41fe8382c82 100644 --- a/tests-integration/tests/export_logical_tables.rs +++ b/tests-integration/tests/export_logical_tables.rs @@ -1453,9 +1453,14 @@ async fn metric_export_v2_cli_resume_roundtrip() { } #[tokio::test] -#[ignore = "requires GT_S3_* credentials and an S3-compatible test bucket"] async fn metric_export_v2_cli_s3_resume_roundtrip() { - metric_export_v2_cli_roundtrip(true).await; + // The CLI takes an explicit `--s3-endpoint`, so this only runs against an + // S3-compatible endpoint (MinIO in CI), not against bucket+region alone. + if std::env::var("GT_S3_ENDPOINT_URL").is_ok_and(|e| !e.is_empty()) + && std::env::var("GT_S3_BUCKET").is_ok_and(|b| !b.is_empty()) + { + metric_export_v2_cli_roundtrip(true).await; + } } async fn metric_export_v2_cli_roundtrip(s3: bool) { diff --git a/tests-integration/tests/flow_ddl_event.rs b/tests-integration/tests/flow_ddl_event.rs index 122c36c84f4..9d2884468d2 100644 --- a/tests-integration/tests/flow_ddl_event.rs +++ b/tests-integration/tests/flow_ddl_event.rs @@ -15,7 +15,9 @@ use std::sync::Arc; use tests_integration::standalone::GreptimeDbStandaloneBuilder; -use tests_integration::test_util::setup_authenticated_grpc_database; +use tests_integration::test_util::{ + setup_authenticated_grpc_database, test_event_recorder_options, +}; use uuid::Uuid; use crate::event_recorder_test_util::{ @@ -31,6 +33,7 @@ const PROCEDURE_ACTOR_PASSWORD: &str = "procedure_actor_pwd"; async fn test_standalone_flow_ddl_events() { common_telemetry::init_default_ut_logging(); let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_flow_ddl_events") + .with_event_recorder_options(test_event_recorder_options()) .build() .await; let instance = standalone.fe_instance(); diff --git a/tests-integration/tests/gc_event.rs b/tests-integration/tests/gc_event.rs index 4721a442a80..48749b784a0 100644 --- a/tests-integration/tests/gc_event.rs +++ b/tests-integration/tests/gc_event.rs @@ -43,6 +43,9 @@ async fn test_batch_gc_event() { common_telemetry::init_default_ut_logging(); let (store_config, _guard) = get_test_store_config(&store_type); let home_dir = create_temp_dir("test_batch_gc_event_data_home"); + // Keeps the production event-recorder flush interval: `assert_sst_count` + // counts SST files across every table, so eagerly flushed event rows would + // show up in the count. let cluster = GreptimeDbClusterBuilder::new("test_batch_gc_event") .await .with_datanodes(1) diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 025d56b959a..970f65a2383 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -19,12 +19,13 @@ use api::v1::greptime_database_client::GreptimeDatabaseClient; use api::v1::greptime_request::Request as RequestBody; use api::v1::greptime_response::Response as ResponseBody; use api::v1::promql_request::Promql; +use api::v1::query_request::Query; use api::v1::value::ValueData; use api::v1::{ AddColumn, AddColumns, AlterTableExpr, Basic, Column, ColumnDataType, ColumnDef, ColumnSchema, CreateTableExpr, GreptimeRequest, InsertRequest, InsertRequests, PromInstantQuery, - PromRangeQuery, PromqlRequest, RequestHeader, Row, RowInsertRequest, RowInsertRequests, Rows, - SemanticType, Value, column, + PromRangeQuery, PromqlRequest, QueryRequest, RequestHeader, Row, RowInsertRequest, + RowInsertRequests, Rows, SemanticType, Value, column, }; use auth::user_provider_from_option; use base64::prelude::{BASE64_STANDARD, Engine as _}; @@ -72,6 +73,7 @@ use tests_integration::test_util::{ setup_grpc_server_with_auto_create_table_disabled, setup_grpc_server_with_user_provider, }; use tonic::Request; +use tonic::codec::CompressionEncoding; use tonic::metadata::MetadataValue; use crate::both_deployment_cases; @@ -89,9 +91,13 @@ macro_rules! grpc_test { async fn [< $test >]() { let store_type = tests_integration::test_util::StorageType::$service; if store_type.test_on() { - let _ = $crate::grpc::$test(store_type).await; + // Support both unit tests and fallible tests without discarding errors. + let result = $crate::grpc::$test(store_type).await; + assert_eq!( + std::process::Termination::report(result), + std::process::ExitCode::SUCCESS, + ); } - } )* } @@ -198,29 +204,128 @@ pub async fn test_grpc_message_size_ok(store_type: StorageType) { let _ = fe_grpc_server.shutdown().await; } +/// Both the server (`servers::grpc::builder`) and the standard client +/// (`configure_tonic_client!`) enable zstd, so a plain round trip never shows +/// whether negotiation actually happened. This drives raw tonic clients instead +/// and asserts on the `grpc-encoding` the server answers with. pub async fn test_grpc_zstd_compression(store_type: StorageType) { - // server and client both support gzip - let config = GrpcServerConfig { - max_recv_message_size: 1024, - max_send_message_size: 1024, - ..Default::default() - }; - let (_db, fe_grpc_server) = setup_grpc_server_with( - store_type, - "test_grpc_zstd_compression", - None, - Some(config), - None, - ) - .await; + let (_db, fe_grpc_server) = setup_grpc_server(store_type, "test_grpc_zstd_compression").await; let addr = fe_grpc_server.bind_addr().unwrap().to_string(); - let grpc_client = Client::with_urls(vec![addr]); + let ddl = |sql: &str| GreptimeRequest { + header: Some(RequestHeader { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + ..Default::default() + }), + request: Some(RequestBody::Query(QueryRequest { + query: Some(Query::Sql(sql.to_string())), + })), + }; + + // Sends zstd and accepts zstd: the server has to decode a compressed request + // body and compress its response. + let mut zstd_client = GreptimeDatabaseClient::connect(format!("http://{addr}")) + .await + .unwrap() + .send_compressed(CompressionEncoding::Zstd) + .accept_compressed(CompressionEncoding::Zstd); + let response = zstd_client + .handle(Request::new(ddl( + "CREATE TABLE zstd_compression (ts TIMESTAMP TIME INDEX, payload STRING)", + ))) + .await + .unwrap(); + assert_eq!( + response + .metadata() + .get("grpc-encoding") + .map(|v| v.to_str().unwrap()), + Some("zstd") + ); + + // A payload far above the zstd frame overhead, so the request body really is + // compressed rather than passed through. + let payload = "compressible-".repeat(4096); + let insert = |ts: i64| GreptimeRequest { + header: Some(RequestHeader { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + ..Default::default() + }), + request: Some(RequestBody::RowInserts(RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: "zstd_compression".to_string(), + rows: Some(Rows { + schema: vec![ + ColumnSchema { + column_name: "ts".to_string(), + semantic_type: SemanticType::Timestamp as i32, + datatype: ColumnDataType::TimestampMillisecond as i32, + ..Default::default() + }, + ColumnSchema { + column_name: "payload".to_string(), + semantic_type: SemanticType::Field as i32, + datatype: ColumnDataType::String as i32, + ..Default::default() + }, + ], + rows: vec![Row { + values: vec![ + Value { + value_data: Some(ValueData::TimestampMillisecondValue(ts)), + }, + Value { + value_data: Some(ValueData::StringValue(payload.clone())), + }, + ], + }], + }), + }], + })), + }; + + let response = zstd_client + .handle(Request::new(insert(1000))) + .await + .unwrap(); + let ResponseBody::AffectedRows(rows) = response.into_inner().response.unwrap(); + assert_eq!(rows.value, 1); + + // A client that does not advertise zstd gets an uncompressed response. + let mut plain_client = GreptimeDatabaseClient::connect(format!("http://{addr}")) + .await + .unwrap(); + let response = plain_client + .handle(Request::new(insert(2000))) + .await + .unwrap(); + assert!(response.metadata().get("grpc-encoding").is_none()); + + // Both payloads survived their respective paths intact. let db = Database::new_with_dbname( format!("{}-{}", DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME), - grpc_client, + Client::with_urls(vec![addr]), ); - db.sql("show tables;").await.unwrap(); + let sql = format!( + "SELECT count(*) AS c FROM zstd_compression WHERE length(payload) = {}", + payload.len() + ); + let OutputData::Stream(stream) = db.sql(&sql).await.unwrap().data else { + panic!("expected a stream"); + }; + let recordbatches = RecordBatches::try_collect(stream).await.unwrap(); + assert_eq!( + recordbatches.pretty_print().unwrap(), + "\ ++---+ +| c | ++---+ +| 2 | ++---+" + ); + let _ = fe_grpc_server.shutdown().await; } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 020e9c023c6..4cc409deec4 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -93,6 +93,31 @@ use yaml_rust::YamlLoader; use crate::both_deployment_cases; use crate::event_recorder_test_util::assert_procedure_actor_by_table; +/// Registers tests that exercise nothing below the HTTP layer, so running them +/// against a remote object store only pays bucket setup and teardown. +#[macro_export] +macro_rules! http_local_tests { + ($($(#[$meta:meta])* $test:ident),*,) => { + mod integration_http_local_test { + $( + #[tokio::test(flavor = "multi_thread")] + $( + #[$meta] + )* + async fn $test() { + let store_type = tests_integration::test_util::StorageType::File; + // Support both unit tests and fallible tests without discarding errors. + let result = $crate::http::$test(store_type).await; + assert_eq!( + std::process::Termination::report(result), + std::process::ExitCode::SUCCESS, + ); + } + )* + } + }; +} + #[macro_export] macro_rules! http_test { ($service:ident, $($(#[$meta:meta])* $test:ident),*,) => { @@ -134,12 +159,7 @@ macro_rules! http_tests { test_prometheus_promql_api, test_prometheus_label_replace_response, test_prom_http_api, - test_metrics_api, - test_health_api, - test_status_api, test_config_api, - test_dynamic_tracer_toggle, - test_dashboard_path, test_dashboard_api, test_prometheus_remote_write, test_prometheus_remote_write_v2, @@ -151,7 +171,6 @@ macro_rules! http_tests { test_vm_proto_remote_write, test_pipeline_api, - test_test_pipeline_api, test_pipeline_name_in_header, test_plain_text_ingestion, test_pipeline_auto_transform, @@ -187,8 +206,6 @@ macro_rules! http_tests { test_loki_json_logs_with_pipeline, test_elasticsearch_logs, test_elasticsearch_logs_with_index, - test_splunk_health, - test_splunk_health_is_public, test_splunk_logs, test_splunk_raw, test_log_query, @@ -1810,15 +1827,21 @@ pub async fn test_splunk_logs(store_type: StorageType) { .await; let client = TestClient::new(app).await; - // Authenticated SQL query (the user-provider harness requires auth on /v1/sql). + // Authenticated SQL statement (the user-provider harness requires auth on + // /v1/sql). Posted as a form so quotes and parentheses survive. async fn query(client: &TestClient, sql: &str) -> String { let res = client - .get(format!("/v1/sql?sql={sql}").as_str()) + .post("/v1/sql") .header("Authorization", basic_auth("greptime_user", "greptime_pwd")) + .form(&[("sql", sql)]) .send() .await; - assert_eq!(res.status(), StatusCode::OK, "query failed: {sql}"); - res.text().await + let status = res.status(); + let body = res.text().await; + assert_eq!(status, StatusCode::OK, "query failed: {sql}: {body}"); + // A failing statement still comes back as 200 with an `error` envelope. + assert!(!body.contains("\"error\""), "query failed: {sql}: {body}"); + body } // HEC `Authorization: Splunk ` + JSON content type. @@ -1969,6 +1992,34 @@ transform: "custom pipeline should have dropped sourcetype (identity would keep it): {create}" ); + // 6b. Ingest into a pre-created table with a fulltext index, then flush: the + // index file is only written on flush, so this is the one path that reads it + // back. `matches_term` is the predicate the index applier recognises. + query( + &client, + "create table splunk_indexed (greptime_timestamp timestamp(9) time index, \ + \"event\" string fulltext index with(backend='bloom'))", + ) + .await; + let res = send_req( + &client, + splunk_headers(), + "/v1/splunk/services/collector/event?table=splunk_indexed", + br#"{"event":"disk pressure on web-01","time":1700000020}"#.to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + query(&client, "ADMIN FLUSH_TABLE('splunk_indexed')").await; + let rows = get_rows_from_output( + &query( + &client, + "select \"event\" from splunk_indexed where matches_term(\"event\", 'pressure')", + ) + .await, + ); + assert_eq!(rows, r#"[["disk pressure on web-01"]]"#); + // 7. Auth failures return HEC codes: missing token -> 2 (401), bad token -> 4 (403). let res = send_req( &client, @@ -2911,6 +2962,18 @@ pub async fn test_prometheus_remote_write(store_type: StorageType) { ) .await; + // A metric-engine logical table stores its rows in the shared physical + // table, so reading them back from SST files goes through a different + // column mapping than the memtable read does. + flush_table(&client, "metric2").await; + validate_data( + "prometheus_remote_write_after_flush", + &client, + "select * from metric2", + table_val, + ) + .await; + // Prom RW tables carry the metric identity; type is inferred from naming. validate_data( "prometheus_remote_write_semantic_identity", @@ -3217,11 +3280,24 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag ) .await; + let histogram_sql = "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;"; + let histogram_rows = "[[3000,{\"count_f64\":null,\"count_i64\":8,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":6.0,\"count_i64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_i64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]"; validate_data( "prometheus_remote_write_v2_native_histogram_rows", &client, - "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;", - "[[3000,{\"count_f64\":null,\"count_i64\":8,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_i64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":6.0,\"count_i64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_i64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]", + histogram_sql, + histogram_rows, + ) + .await; + + // The native histogram column is a struct of nested lists, so its SST + // encoding differs from its memtable representation. + flush_table(&client, "remote_write_v2_latency_seconds").await; + validate_data( + "prometheus_remote_write_v2_native_histogram_rows_after_flush", + &client, + histogram_sql, + histogram_rows, ) .await; @@ -4283,6 +4359,31 @@ transform: ) .await; + // The inverted, skipping and fulltext index files are only written on flush, + // so the queries below are the first thing that reads them back. + flush_table(&client, "test_db.logs1").await; + validate_data( + "pipeline_db_inverted_index", + &client, + "select id2 from test_db.logs1 where id1 = 2436", + "[[2528]]", + ) + .await; + validate_data( + "pipeline_db_skipping_index", + &client, + "select id1 from test_db.logs1 where type = 'I'", + "[[2436]]", + ) + .await; + validate_data( + "pipeline_db_fulltext_index", + &client, + "select id1 from test_db.logs1 where matches(log, 'ClusterAdapter')", + "[[2436]]", + ) + .await; + // 5. remove pipeline let encoded_ver_str: String = url::form_urlencoded::byte_serialize(version_str.as_bytes()).collect(); @@ -5079,6 +5180,24 @@ transform: ) .await; + // The fulltext and skipping index files are only written on flush, so the + // queries below are the first thing that reads them back. + flush_table(&client, "pipeline_index_options").await; + validate_data( + "pipeline_index_options_fulltext", + &client, + "select trace_id from pipeline_index_options where matches(message, 'greptime')", + "[[42]]", + ) + .await; + validate_data( + "pipeline_index_options_skipping", + &client, + "select message from pipeline_index_options where trace_id = 42", + r#"[["hello greptime"]]"#, + ) + .await; + guard.remove_all().await; } @@ -6995,6 +7114,18 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { ) .await; + // A metric-engine logical table stores its rows in the shared physical + // table, so reading them back from SST files goes through a different + // column mapping than the memtable read does. + flush_table(&client, "claude_code_cost_usage_USD_total").await; + validate_data( + "otlp_metrics_all_select_after_flush", + &client, + "select * from `claude_code_cost_usage_USD_total` order by model desc;", + expected, + ) + .await; + // The synthesized resource descriptor: a plain mito info table keyed by // the raw OTel attribute keys (service.name is the only allowlisted // resource attribute in this payload), independent of the promote/ignore @@ -7846,6 +7977,17 @@ pub async fn test_otlp_traces_v0(store_type: StorageType) { ) .await; + // The attribute columns are JSON and the event/link columns are lists, whose + // SST encodings differ from their memtable representations. + flush_table(&client, "opentelemetry_traces").await; + validate_data( + "otlp_traces_after_flush", + &client, + "select * from opentelemetry_traces;", + expected, + ) + .await; + guard.remove_all().await; } @@ -7972,6 +8114,30 @@ pub(crate) async fn test_otlp_traces_v2( ) .await; + // JSON2 path extraction has to resolve the same way against SST files as it + // does against the memtable. + flush_table(&client, "trace_v2_spans").await; + validate_data( + "otlp_traces_v2_json2_after_flush", + &client, + "select service_name, span_attributes.\"http.status_code\"::BIGINT, \ + resource_attributes.\"deployment.environment\"::STRING, \ + scope_attributes.enabled::BOOLEAN from trace_v2_spans;", + r#"[["frontend",200,"production",true]]"#, + ) + .await; + validate_data( + "otlp_traces_v2_events_links_after_flush", + &client, + r#"select json_get_string(span_events, '$[0].name'), + json_get_int(span_events, '$[0].attributes."event.code"'), + json_get_string(span_links, '$[0].trace_id'), + json_get_string(span_links, '$[0].attributes."link.type"') + from trace_v2_spans;"#, + r#"[["cache.hit",7,"cc9e0991a2e63d274984bd44ee669203","follows_from"]]"#, + ) + .await; + guard.remove_all().await; Ok(()) } @@ -8125,6 +8291,17 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { let expected = r#"[[1736480942444376000,1736480942444499000,123000,null,"c05d7a4ec8e1f231f02ed6e8da8655b4","d24f921c75f68e23","SPAN_KIND_CLIENT","lets-go","STATUS_CODE_UNSET","","","telemetrygen","","telemetrygen","1.2.3.4","telemetrygen-server",[],[]],[1736480942444376000,1736480942444499000,123000,"d24f921c75f68e23","c05d7a4ec8e1f231f02ed6e8da8655b4","9630f2916e2f7909","SPAN_KIND_SERVER","okey-dokey-0","STATUS_CODE_UNSET","","","telemetrygen","","telemetrygen","1.2.3.4","telemetrygen-client",[],[]],[1736480942444589000,1736480942444712000,123000,null,"cc9e0991a2e63d274984bd44ee669203","eba7be77e3558179","SPAN_KIND_CLIENT","lets-go","STATUS_CODE_UNSET","","","telemetrygen","","telemetrygen","1.2.3.4","telemetrygen-server",[],[]],[1736480942444589000,1736480942444712000,123000,"eba7be77e3558179","cc9e0991a2e63d274984bd44ee669203","8f847259b0f6e1ab","SPAN_KIND_SERVER","okey-dokey-0","STATUS_CODE_UNSET","","","telemetrygen","","telemetrygen","1.2.3.4","telemetrygen-client",[],[]]]"#; validate_data("otlp_traces", &client, "select * from mytable;", expected).await; + // `span_events` / `span_links` are JSON columns and the table is partitioned + // on `trace_id`, so the read back has to reassemble SST files across regions. + flush_table(&client, "mytable").await; + validate_data( + "otlp_traces_after_flush", + &client, + "select * from mytable;", + expected, + ) + .await; + // The trace v1 main table carries the trace identity (events/links preserved as // JSON columns by the v1 model). validate_data( @@ -8961,6 +9138,17 @@ pub async fn test_otlp_logs(store_type: StorageType) { ) .await; + // The attribute columns are JSON, whose SST encoding differs from their + // memtable representation. + flush_table(&client, "opentelemetry_logs").await; + validate_data( + "otlp_logs_after_flush", + &client, + "select * from opentelemetry_logs;", + expected, + ) + .await; + // The auto-created log table carries the log identity. validate_data( "otlp_logs_semantic_identity", @@ -9343,6 +9531,17 @@ pub async fn test_loki_pb_logs(store_type: StorageType) { ) .await; + // `structured_metadata` is a JSON column, whose SST encoding differs from its + // memtable representation. + flush_table(&client, "loki_table_name").await; + validate_data( + "loki_pb_content_after_flush", + &client, + "select * from loki_table_name;", + expected, + ) + .await; + guard.remove_all().await; } @@ -9548,6 +9747,17 @@ pub async fn test_loki_json_logs(store_type: StorageType) { ) .await; + // `structured_metadata` is a JSON column, whose SST encoding differs from its + // memtable representation. + flush_table(&client, "loki_table_name").await; + validate_data( + "loki_json_content_after_flush", + &client, + "select * from loki_table_name;", + expected, + ) + .await; + guard.remove_all().await; } @@ -9767,7 +9977,7 @@ pub async fn test_log_query(store_type: StorageType) { // prepare data with SQL API let res = client - .get("/v1/sql?sql=create table logs (`ts` timestamp time index, `message` string);") + .get("/v1/sql?sql=create table logs (`ts` timestamp time index, `message` string fulltext index with(backend='bloom'));") .send() .await; assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); @@ -9842,6 +10052,35 @@ pub async fn test_log_query(store_type: StorageType) { ] ); + // The fulltext index file is only written on flush. Replay the log query to + // cover the SST read path, and add a `matches_term` query, which is the + // predicate the index applier actually recognises. + flush_table(&client, "logs").await; + let res = client + .post("/v1/logs") + .header("Content-Type", "application/json") + .body(serde_json::to_string(&log_query).unwrap()) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await); + let resp = res.text().await; + let output = serde_json::from_str::(&resp).unwrap(); + let rows = output["output"][0]["records"]["rows"].as_array().unwrap(); + let mut flushed_messages = rows + .iter() + .map(|row| row[1].as_str().unwrap()) + .collect::>(); + flushed_messages.sort_unstable(); + assert_eq!(flushed_messages, messages); + + validate_data( + "log_query_fulltext_index", + &client, + "select message from logs where matches_term(message, 'at-explicit-end')", + r#"[["at-explicit-end"]]"#, + ) + .await; + let res = client .get("/v1/sql?sql=create table logs_limit (`ts` timestamp time index, `message` string);") .send() @@ -10865,11 +11104,20 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let mut req: ExportTraceServiceRequest = serde_json::from_str(content).unwrap(); - // Modify timestamp fields + // The fixture is written against a fixed base instant. Move it next to now: + // this table is created with `ttl=7d`, and stale spans would be dropped as + // expired the moment they are flushed into SST files. let now = Utc::now().timestamp_nanos_opt().unwrap() as u64; + let delta_micros = (now / 1_000) as i64 - 60_000_000 - JAEGER_V1_FIXTURE_BASE_MICROS as i64; + let delta_nanos = delta_micros * 1_000; for span in req.resource_spans.iter_mut() { for scope_span in span.scope_spans.iter_mut() { - // Only modify the timestamp fields for the span with the name "test-jaeger-get-operations" to current time. + for span in scope_span.spans.iter_mut() { + span.start_time_unix_nano = (span.start_time_unix_nano as i64 + delta_nanos) as u64; + span.end_time_unix_nano = (span.end_time_unix_nano as i64 + delta_nanos) as u64; + } + // `/api/operations` is queried without a time range, so these spans + // stay pinned to the last few seconds. if scope_span.scope.as_ref().unwrap().name == "test-jaeger-get-operations" { for span in scope_span.spans.iter_mut() { span.start_time_unix_nano = now - 5_000_000_000; // 5 seconds ago @@ -10946,7 +11194,8 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { } "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/operations` API. @@ -10995,12 +11244,13 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { } "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/services/{service_name}/operations` API. let res = client - .get("/v1/jaeger/api/services/test-jaeger-query-api/operations?start=1738726754492421&end=1738726754642422") + .get(&shift_fixture_micros("/v1/jaeger/api/services/test-jaeger-query-api/operations?start=1738726754492421&end=1738726754642422", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11023,7 +11273,8 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { } "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/traces/{trace_id}` API without start and end. @@ -11149,12 +11400,26 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); + assert_eq!(resp, expected); + + // Replay the same lookup against SST files: `trace_id` and `service_name` + // carry BLOOM skipping indexes whose puffin files only exist after a flush, + // and `span_events` / `span_links` are JSON columns. + flush_table(&client, trace_table_name).await; + let res = client + .get("/v1/jaeger/api/traces/5611dce1bc9ebed65352d99a027b08ea") + .header("x-greptime-trace-table-name", trace_table_name) + .send() + .await; + assert_eq!(StatusCode::OK, res.status()); + let resp: Value = serde_json::from_str(&res.text().await).unwrap(); assert_eq!(resp, expected); // Test `/api/traces/{trace_id}` API with start and end in microseconds. let res = client - .get("/v1/jaeger/api/traces/5611dce1bc9ebed65352d99a027b08ea?start=1738726754492421&end=1738726754642422") + .get(&shift_fixture_micros("/v1/jaeger/api/traces/5611dce1bc9ebed65352d99a027b08ea?start=1738726754492421&end=1738726754642422", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11275,12 +11540,16 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/traces/{trace_id}` API for non-existent trace. let res = client - .get("/v1/jaeger/api/traces/0000000000000000000000000000dead") + .get(&shift_fixture_micros( + "/v1/jaeger/api/traces/0000000000000000000000000000dead", + delta_micros, + )) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11299,12 +11568,13 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { } "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/traces` API. let res = client - .get("/v1/jaeger/api/traces?service=test-jaeger-query-api&operation=access-mysql&start=1738726754492421&end=1738726754642422&tags=%7B%22operation.type%22%3A%22access-mysql%22%7D") + .get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&operation=access-mysql&start=1738726754492421&end=1738726754642422&tags=%7B%22operation.type%22%3A%22access-mysql%22%7D", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11426,13 +11696,14 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/traces` API with tags. // 1. first query without tags, get 2 results let res = client - .get("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754492422&end=1738726754592422") + .get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754492422&end=1738726754592422", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11443,12 +11714,13 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // 2. second query with tags, get 1 result let res = client -.get("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754492422&end=1738726754592422&tags=%7B%22operation.type%22%3A%22access-pg%22%7D") +.get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754492422&end=1738726754592422&tags=%7B%22operation.type%22%3A%22access-pg%22%7D", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11459,14 +11731,15 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { "#; let resp: Value = serde_json::from_str(&res.text().await).unwrap(); - let expected: Value = serde_json::from_str(expected).unwrap(); + let expected: Value = + serde_json::from_str(&shift_fixture_micros(expected, delta_micros)).unwrap(); assert_eq!(resp, expected); // Test `/api/traces` API with Grafana User-Agent. // When user agent is Grafana, only return at most 3 spans per trace (earliest by timestamp). // Trace `5611dce1bc9ebed65352d99a027b08fb` has 4 spans, so only 3 should be returned. let res = client - .get("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004") + .get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .header("User-Agent", "Grafana/8.0.0") .send() @@ -11514,7 +11787,7 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { // Test `/api/traces` API without User-Agent (default behavior). // All 4 spans should be returned for the trace. let res = client - .get("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004") + .get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .send() .await; @@ -11534,7 +11807,7 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { // Test `/api/traces` API with Jaeger User-Agent (should return all spans like default). let res = client - .get("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004") + .get(&shift_fixture_micros("/v1/jaeger/api/traces?service=test-jaeger-query-api&start=1738726754600000&end=1738726754700004", delta_micros)) .header("x-greptime-trace-table-name", trace_table_name) .header("User-Agent", "Jaeger-Query/1.0.0") .send() @@ -11613,6 +11886,59 @@ pub async fn test_influxdb_write(store_type: StorageType) { guard.remove_all().await; } +/// Flushes `table` so the rows written so far move from the memtable into SST +/// files. Replaying a query afterwards covers the encode/decode round trip and, +/// for indexed columns, the puffin files that only exist after a flush. +/// Base microsecond instant the `test_jaeger_query_api_for_trace_v1` fixture and +/// its expected responses are written against. +const JAEGER_V1_FIXTURE_BASE_MICROS: u64 = 1_738_726_754_492_421; + +/// Shifts every fixture timestamp in `text` by `delta` microseconds, so expected +/// responses and query bounds follow the payload after it is moved next to now. +/// +/// Only integer literals within a day of [`JAEGER_V1_FIXTURE_BASE_MICROS`] are +/// rewritten; durations, counts and ids are left alone. +fn shift_fixture_micros(text: &str, delta: i64) -> String { + let mut out = String::with_capacity(text.len()); + let mut rest = text; + while let Some(start) = rest.find(|c: char| c.is_ascii_digit()) { + out.push_str(&rest[..start]); + let digits_end = rest[start..] + .find(|c: char| !c.is_ascii_digit()) + .map_or(rest.len(), |offset| start + offset); + let digits = &rest[start..digits_end]; + match digits.parse::() { + Ok(value) if (value - JAEGER_V1_FIXTURE_BASE_MICROS as i64).abs() < 86_400_000_000 => { + out.push_str(&(value + delta).to_string()) + } + _ => out.push_str(digits), + } + rest = &rest[digits_end..]; + } + out.push_str(rest); + out +} + +async fn flush_table(client: &TestClient, table: &str) { + // The argument is parsed as a SQL table name, so an unquoted identifier is + // lower-cased and will not resolve a mixed-case table. + let quoted = table + .split('.') + .map(|part| format!("\"{part}\"")) + .collect::>() + .join("."); + let res = client + .post("/v1/sql") + .form(&[("sql", format!("ADMIN FLUSH_TABLE('{quoted}')"))]) + .send() + .await; + let status = res.status(); + let body = res.text().await; + assert_eq!(status, StatusCode::OK, "flush {table} fail: {body}"); + // A failing statement still comes back as 200 with an `error` envelope. + assert!(!body.contains("\"error\""), "flush {table} fail: {body}"); +} + async fn validate_data(test_name: &str, client: &TestClient, sql: &str, expected: &str) { let res = client .get(format!("/v1/sql?sql={sql}").as_str()) diff --git a/tests-integration/tests/main.rs b/tests-integration/tests/main.rs index b61b6cd3e19..278a7ba3851 100644 --- a/tests-integration/tests/main.rs +++ b/tests-integration/tests/main.rs @@ -53,6 +53,17 @@ grpc_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs); http_tests!(File, S3, S3WithCache, Oss, Azblob, Gcs); +http_local_tests!( + test_health_api, + test_status_api, + test_metrics_api, + test_splunk_health, + test_splunk_health_is_public, + test_dynamic_tracer_toggle, + test_dashboard_path, + test_test_pipeline_api, +); + #[tokio::test(flavor = "multi_thread")] async fn test_http_auth_from_standalone_user_provider_config() { http::test_http_auth_from_standalone_user_provider_config().await; diff --git a/tests-integration/tests/reconciliation_event.rs b/tests-integration/tests/reconciliation_event.rs index 19eaea6ff2c..a366217a7d9 100644 --- a/tests-integration/tests/reconciliation_event.rs +++ b/tests-integration/tests/reconciliation_event.rs @@ -20,6 +20,7 @@ use servers::query_handler::sql::SqlQueryHandler; use session::context::QueryContext; use table::table_reference::TableReference; use tests_integration::cluster::GreptimeDbClusterBuilder; +use tests_integration::test_util::test_event_recorder_options; use crate::event_recorder_test_util::{assert_eventually_eq, find_eventually_string}; @@ -33,6 +34,7 @@ async fn test_catalog_and_database_reconciliation_events() { let cluster = GreptimeDbClusterBuilder::new("catalog_database_reconciliation_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .build(true) .await; @@ -243,6 +245,7 @@ async fn test_table_reconciliation_events() { let cluster = GreptimeDbClusterBuilder::new("table_reconciliation_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .build(true) .await; @@ -351,6 +354,7 @@ async fn test_logical_table_reconciliation_events() { let cluster = GreptimeDbClusterBuilder::new("logical_table_reconciliation_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .build(true) .await; diff --git a/tests-integration/tests/repartition_event.rs b/tests-integration/tests/repartition_event.rs index 906eb3cfee3..aaefcc2d49c 100644 --- a/tests-integration/tests/repartition_event.rs +++ b/tests-integration/tests/repartition_event.rs @@ -22,6 +22,7 @@ use mito2::gc::GcConfig; use tests_integration::cluster::GreptimeDbClusterBuilder; use tests_integration::test_util::{ StorageType, get_test_store_config, setup_authenticated_grpc_database, + test_event_recorder_options, }; use crate::event_recorder_test_util::{ @@ -44,6 +45,7 @@ async fn test_repartition_event() { let home_dir = create_temp_dir("test_repartition_event_data_home"); let cluster = GreptimeDbClusterBuilder::new("test_repartition_event") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .with_store_config(store_config) .with_shared_home_dir(Arc::new(home_dir)) diff --git a/tests-integration/tests/table_ddl_event.rs b/tests-integration/tests/table_ddl_event.rs index 8bad66dae3c..6cce1736151 100644 --- a/tests-integration/tests/table_ddl_event.rs +++ b/tests-integration/tests/table_ddl_event.rs @@ -41,7 +41,9 @@ use servers::query_handler::grpc::GrpcQueryHandler; use servers::query_handler::sql::SqlQueryHandler; use session::context::{Channel, QueryContext}; use tests_integration::cluster::GreptimeDbClusterBuilder; -use tests_integration::test_util::setup_authenticated_grpc_database; +use tests_integration::test_util::{ + setup_authenticated_grpc_database, test_event_recorder_options, +}; use crate::event_recorder_test_util::{ assert_eventually_eq, assert_procedure_actor, assert_single_event, find_eventually_string, @@ -79,6 +81,7 @@ async fn test_table_ddl_procedure_events() { } let cluster = GreptimeDbClusterBuilder::new("table_ddl_procedure_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .with_metasrv_gc_config(gc_options) .with_datanode_gc_config(GcConfig { diff --git a/tests-integration/tests/view_ddl_event.rs b/tests-integration/tests/view_ddl_event.rs index b21838995c3..8d17e9b4e25 100644 --- a/tests-integration/tests/view_ddl_event.rs +++ b/tests-integration/tests/view_ddl_event.rs @@ -17,9 +17,9 @@ use std::sync::Arc; use common_test_util::temp_dir::create_temp_dir; use common_wal::config::DatanodeWalConfig; use tests_integration::cluster::GreptimeDbClusterBuilder; -use tests_integration::standalone::GreptimeDbStandaloneBuilder; use tests_integration::test_util::{ StorageType, get_test_store_config, setup_authenticated_grpc_database, + test_event_recorder_options, }; use uuid::Uuid; @@ -44,6 +44,7 @@ async fn test_view_ddl_events() { let home_dir = create_temp_dir("test_view_ddl_events_data_home"); let cluster = GreptimeDbClusterBuilder::new("test_view_ddl_events") .await + .with_event_recorder_options(test_event_recorder_options()) .with_datanodes(1) .with_store_config(store_config) .with_datanode_wal_config(DatanodeWalConfig::Noop) @@ -58,19 +59,6 @@ async fn test_view_ddl_events() { execute_view_ddl(instance, &source_table, &view).await; } -#[tokio::test(flavor = "multi_thread")] -async fn test_standalone_view_ddl_events() { - common_telemetry::init_default_ut_logging(); - let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_view_ddl_events") - .build() - .await; - let suffix = Uuid::new_v4().simple(); - let source_table = format!("view_ddl_event_source_{suffix}"); - let view = format!("view_ddl_event_{suffix}"); - - execute_view_ddl(standalone.fe_instance(), &source_table, &view).await; -} - async fn execute_view_ddl( instance: &Arc, source_table: &str, diff --git a/tests-integration/tests/wal_prune_event.rs b/tests-integration/tests/wal_prune_event.rs index db86ade4d9d..3879fb3e653 100644 --- a/tests-integration/tests/wal_prune_event.rs +++ b/tests-integration/tests/wal_prune_event.rs @@ -26,6 +26,7 @@ use meta_srv::procedure::wal_prune::{Context as WalPruneContext, WalPruneProcedu use rskafka::client::partition::{Compression, UnknownTopicHandling}; use rskafka::record::Record; use tests_integration::standalone::GreptimeDbStandaloneBuilder; +use tests_integration::test_util::test_event_recorder_options; use crate::event_recorder_test_util::assert_single_event; @@ -35,6 +36,7 @@ async fn test_standalone_wal_prune_event() { common_telemetry::init_default_ut_logging(); let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_wal_prune_event") + .with_event_recorder_options(test_event_recorder_options()) .build() .await; let topic_name = format!("test_standalone_wal_prune_event-{}", uuid::Uuid::new_v4()); From 0f118bc5ba3ad591824562f3ede4b8fd536fb86d Mon Sep 17 00:00:00 2001 From: Ning Sun Date: Wed, 23 Sep 2026 01:55:46 +0000 Subject: [PATCH 2/2] fix: serialize struct to json in postgres (#9170) * feat: serialize struct to json in postgres * fix: support view scalars and preserve null structs in scalar-to-value conversion Address PR review: - Utf8View/BinaryView ScalarValues now convert like their non-view forms instead of failing row extraction for struct columns - a null struct scalar converts to Value::Null so a null struct inside a list stays null in the serialized JSON Signed-off-by: Ning Sun * fix: return errors instead of panics for unsupported arrow field types Struct-typed query results with arrow field types greptimedb cannot represent (e.g. Decimal256) used to panic during schema conversion and row extraction, dropping the client connection. They now surface as query errors: - ConcreteDataType::try_from builds struct types fallibly via the new StructType::try_from_arrow_fields - Value::try_from(ScalarValue::Struct) uses the same fallible path - new try_value_from_array converts an arrow element to Value with error propagation, used by the postgres struct encoding Signed-off-by: Ning Sun --------- Signed-off-by: Ning Sun --- src/datatypes/src/data_type.rs | 4 +- src/datatypes/src/types/struct_type.rs | 19 ++ src/datatypes/src/value.rs | 99 +++++++- src/servers/src/postgres/types.rs | 315 ++++++++++++++++++++++++- tests-integration/src/test_util.rs | 55 +++++ tests-integration/tests/sql.rs | 233 +++++++++++++++++- 6 files changed, 711 insertions(+), 14 deletions(-) diff --git a/src/datatypes/src/data_type.rs b/src/datatypes/src/data_type.rs index f2ed05534d6..8ad388c2ff7 100644 --- a/src/datatypes/src/data_type.rs +++ b/src/datatypes/src/data_type.rs @@ -492,7 +492,9 @@ impl TryFrom<&ArrowDataType> for ConcreteDataType { ArrowDataType::Decimal128(precision, scale) => { ConcreteDataType::decimal128_datatype(*precision, *scale) } - ArrowDataType::Struct(fields) => ConcreteDataType::Struct(StructType::from(fields)), + ArrowDataType::Struct(fields) => { + ConcreteDataType::Struct(StructType::try_from_arrow_fields(fields)?) + } ArrowDataType::Float16 | ArrowDataType::Date64 | ArrowDataType::FixedSizeBinary(_) diff --git a/src/datatypes/src/types/struct_type.rs b/src/datatypes/src/types/struct_type.rs index ffd1d1c0b25..fdbfd4886b9 100644 --- a/src/datatypes/src/types/struct_type.rs +++ b/src/datatypes/src/types/struct_type.rs @@ -19,6 +19,7 @@ use arrow::datatypes::{DataType as ArrowDataType, Field}; use arrow_schema::Fields; use serde::{Deserialize, Serialize}; +use crate::error::Result; use crate::prelude::{ConcreteDataType, DataType, LogicalTypeId}; use crate::value::Value; use crate::vectors::StructVectorBuilder; @@ -100,6 +101,24 @@ impl StructType { } } + /// Like the `From<&Fields>` impl, but returns an error for arrow field + /// types greptimedb cannot represent instead of panicking. + pub fn try_from_arrow_fields(fields: &Fields) -> Result { + let struct_fields = fields + .iter() + .map(|field| { + Ok(StructField::new( + field.name().clone(), + ConcreteDataType::try_from(field.data_type())?, + field.is_nullable(), + )) + }) + .collect::>>()?; + Ok(StructType { + fields: Arc::new(struct_fields), + }) + } + pub fn fields(&self) -> Arc> { self.fields.clone() } diff --git a/src/datatypes/src/value.rs b/src/datatypes/src/value.rs index 56a551484f2..da1df882aba 100644 --- a/src/datatypes/src/value.rs +++ b/src/datatypes/src/value.rs @@ -1152,12 +1152,13 @@ impl TryFrom for Value { ScalarValue::UInt16(u) => Value::from(u), ScalarValue::UInt32(u) => Value::from(u), ScalarValue::UInt64(u) => Value::from(u), - ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) => { + ScalarValue::Utf8(s) | ScalarValue::LargeUtf8(s) | ScalarValue::Utf8View(s) => { Value::from(s.map(StringBytes::from)) } ScalarValue::Binary(b) | ScalarValue::LargeBinary(b) - | ScalarValue::FixedSizeBinary(_, b) => Value::from(b.map(Bytes::from)), + | ScalarValue::FixedSizeBinary(_, b) + | ScalarValue::BinaryView(b) => Value::from(b.map(Bytes::from)), ScalarValue::List(array) => { // this is for item type let datatype = ConcreteDataType::try_from(&array.value_type())?; @@ -1221,7 +1222,15 @@ impl TryFrom for Value { .map(|v| Value::Decimal128(Decimal128::new(v, p, s))) .unwrap_or(Value::Null), ScalarValue::Struct(struct_array) => { - let struct_type = StructType::from(struct_array.fields()); + // A struct scalar carries a single element; a null (or empty) + // element means the struct itself is null, for example a null + // struct inside a list. + if struct_array.is_empty() || struct_array.is_null(0) { + return Ok(Value::Null); + } + // Build the struct type fallibly: an unrepresentable arrow + // field type must surface as an error, not a panic. + let struct_type = StructType::try_from_arrow_fields(struct_array.fields())?; let items = struct_array .columns() .iter() @@ -1244,8 +1253,6 @@ impl TryFrom for Value { | ScalarValue::LargeListView(_) | ScalarValue::Union(_, _, _) | ScalarValue::Float16(_) - | ScalarValue::Utf8View(_) - | ScalarValue::BinaryView(_) | ScalarValue::Map(_) | ScalarValue::Date64(_) | ScalarValue::RunEndEncoded(_, _, _) => { @@ -1259,6 +1266,21 @@ impl TryFrom for Value { } } +/// Converts the value at `index` of an arrow array into a [`Value`]. +/// +/// Unlike the unchecked conversion in [`crate::vectors::StructVector`]'s `get`, +/// this propagates conversion errors (for example arrow types greptimedb +/// cannot represent, such as `Decimal256`) instead of panicking, so callers +/// can surface a proper error to the user. +pub fn try_value_from_array(array: &dyn Array, index: usize) -> Result { + if array.is_null(index) { + return Ok(Value::Null); + } + let scalar = + ScalarValue::try_from_array(array, index).context(ConvertArrowArrayToScalarsSnafu)?; + Value::try_from(scalar) +} + impl From> for Value { fn from(value: ValueRef<'_>) -> Self { match value { @@ -2157,6 +2179,73 @@ pub(crate) mod tests { Value::Struct(struct_value), scalar_struct_value.try_into().unwrap() ); + + // view-typed strings and binaries convert like their non-view forms + assert_eq!( + Value::String("abc".into()), + ScalarValue::Utf8View(Some("abc".into())) + .try_into() + .unwrap() + ); + assert_eq!(Value::Null, ScalarValue::Utf8View(None).try_into().unwrap()); + assert_eq!( + Value::Binary(Bytes::from(vec![1, 2])), + ScalarValue::BinaryView(Some(vec![1, 2])) + .try_into() + .unwrap() + ); + assert_eq!( + Value::Null, + ScalarValue::BinaryView(None).try_into().unwrap() + ); + + // a null struct scalar is a null value, not a struct of null fields + let null_struct = ScalarStructBuilder::new_null(build_struct_type().as_arrow_fields()); + assert_eq!(Value::Null, null_struct.try_into().unwrap()); + } + + #[test] + fn test_try_value_from_array() { + use arrow_array::Int32Array; + use datafusion_common::arrow::datatypes::i256; + + let array = Int32Array::from(vec![Some(1), None]); + assert_eq!(Value::Int32(1), try_value_from_array(&array, 0).unwrap()); + assert_eq!(Value::Null, try_value_from_array(&array, 1).unwrap()); + + // a struct with representable fields converts + let supported = ScalarStructBuilder::new() + .with_name_and_scalar("i", ScalarValue::Int32(Some(7))) + .build() + .unwrap(); + let ScalarValue::Struct(array) = supported else { + unreachable!(); + }; + assert_eq!( + Value::Struct(StructValue::new( + vec![Value::Int32(7)], + StructType::new(std::sync::Arc::new(vec![StructField::new( + "i", + ConcreteDataType::int32_datatype(), + false, + )])) + )), + try_value_from_array(array.as_ref(), 0).unwrap() + ); + + // an unsupported arrow field type errors instead of panicking + let unsupported = ScalarStructBuilder::new() + .with_name_and_scalar( + "d", + ScalarValue::Decimal256(Some(i256::from_i128(1)), 38, 10), + ) + .build() + .unwrap(); + let ScalarValue::Struct(array) = unsupported else { + unreachable!(); + }; + let error = try_value_from_array(array.as_ref(), 0).unwrap_err(); + assert!(error.to_string().contains("Unsupported arrow data type")); } #[test] diff --git a/src/servers/src/postgres/types.rs b/src/servers/src/postgres/types.rs index 033a1c665e2..c83debd39a1 100644 --- a/src/servers/src/postgres/types.rs +++ b/src/servers/src/postgres/types.rs @@ -19,7 +19,7 @@ use std::pin::Pin; use std::sync::{Arc, LazyLock}; use std::task::{Context, Poll}; -use arrow::array::{Array, AsArray}; +use arrow::array::{Array, ArrayRef, AsArray}; use arrow_pg::encoder::{Encoder, encode_value}; use arrow_pg::list_encoder::encode_list; use arrow_schema::{DataType, TimeUnit}; @@ -37,7 +37,7 @@ use datatypes::json::JsonSettings; use datatypes::prelude::{ConcreteDataType, DataType as _, Value}; use datatypes::schema::{Schema, SchemaRef}; use datatypes::types::{Decimal128Type, IntervalType, TimestampType, jsonb_to_string}; -use datatypes::value::StructValue; +use datatypes::value::try_value_from_array; use futures::Stream; use pg_interval::Interval as PgInterval; use pgwire::api::Type; @@ -234,7 +234,8 @@ impl ToSqlText for OidAliasValue<'_> { } } -/// this function will encode greptime's `StructValue` into PostgreSQL jsonb type +/// this function will encode greptime's structured values (`StructValue`, or a +/// list of them) into PostgreSQL jsonb type /// /// Note that greptimedb has different types of StructValue for storing json data, /// based on policy defined in `JsonSettings`. But here the `StructValue` @@ -245,18 +246,35 @@ impl ToSqlText for OidAliasValue<'_> { /// arrays: element in array must be the same type fn encode_struct( _query_ctx: &QueryContextRef, - struct_value: StructValue, + value: Value, builder: &mut S, pg_field: &FieldInfo, ) -> PgWireResult<()> { let encoding_setting = JsonSettings::default(); let json_value = encoding_setting - .decode(Value::Struct(struct_value)) + .decode(value) .map_err(|e| PgWireError::ApiError(Box::new(e)))?; builder.encode_field(&json_value, pg_field) } +/// Encodes one row of a structured column (`Struct`, or `List` of `Struct`) as +/// a PostgreSQL `json` value. +/// +/// Returns an error for arrow field types greptimedb cannot represent +/// (e.g. `Decimal256`) instead of panicking, so the client receives a proper +/// error response. +fn encode_structured_json( + query_ctx: &QueryContextRef, + column: &ArrayRef, + i: usize, + encoder: &mut S, + pg_field: &FieldInfo, +) -> PgWireResult<()> { + let value = try_value_from_array(column.as_ref(), i).map_err(convert_err)?; + encode_struct(query_ctx, value, encoder, pg_field) +} + pub(crate) struct RecordBatchRowStream where S: Encoder, @@ -369,6 +387,11 @@ where } } + DataType::List(list_field) + if matches!(list_field.data_type(), DataType::Struct(_)) => + { + encode_structured_json(query_ctx, column, i, encoder, pg_field)?; + } DataType::List(_) => { let array = column.as_list::(); let items = array.value(i); @@ -376,7 +399,7 @@ where encode_list(encoder, items, pg_field)?; } DataType::Struct(_) => { - encode_struct(query_ctx, Default::default(), encoder, pg_field)?; + encode_structured_json(query_ctx, column, i, encoder, pg_field)?; } DataType::Utf8 => { let arrow_field = arrow_schema.field(j); @@ -445,7 +468,9 @@ pub(super) fn type_gt_to_pg(origin: &ConcreteDataType) -> Result { &ConcreteDataType::Decimal128(_) => Ok(Type::NUMERIC_ARRAY), &ConcreteDataType::Json(_) => Ok(Type::JSON_ARRAY), &ConcreteDataType::Duration(_) => Ok(Type::INTERVAL_ARRAY), - &ConcreteDataType::Struct(_) => Ok(Type::JSON_ARRAY), + // a list of structs is encoded as a single JSON array, matching the + // struct-to-json mapping above + &ConcreteDataType::Struct(_) => Ok(Type::JSON), &ConcreteDataType::Dictionary(_) | &ConcreteDataType::Vector(_) | &ConcreteDataType::List(_) => server_error::UnsupportedDataTypeSnafu { @@ -1921,6 +1946,282 @@ mod test { } } + #[test] + fn test_encode_list_of_struct_as_json() { + use datatypes::types::{StructField, StructType}; + use datatypes::value::{ListValue, StructValue}; + use datatypes::vectors::Helper; + use pgwire::messages::data::DataRow; + + // struct type: {a: int32, b: float64} + let fields = vec![ + StructField::new("a".to_string(), ConcreteDataType::int32_datatype(), true), + StructField::new("b".to_string(), ConcreteDataType::float64_datatype(), true), + ]; + let struct_type = + ConcreteDataType::struct_datatype(StructType::new(Arc::new(fields.clone()))); + let list_type = ConcreteDataType::list_datatype(Arc::new(struct_type.clone())); + + let make_struct = |a: i32, b: f64| { + Value::Struct(StructValue::new( + vec![Value::Int32(a), Value::Float64(b.into())], + StructType::new(Arc::new(fields.clone())), + )) + }; + let rows = vec![ + Value::List(ListValue::new( + vec![make_struct(1, 1.5), Value::Null, make_struct(3, 3.5)], + Arc::new(struct_type.clone()), + )), + Value::List(ListValue::new(vec![], Arc::new(struct_type.clone()))), + Value::Null, + ]; + let vector = Helper::try_from_row_into_vector(&rows, &list_type).unwrap(); + + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "l".to_string(), + list_type.clone(), + true, + )])); + let record_batch = RecordBatch::new(schema.clone(), vec![vector]).unwrap(); + + // The list-of-struct column is declared as a single json value. + let pg_schema = Arc::new(vec![FieldInfo::new( + "l".into(), + None, + None, + type_gt_to_pg(&list_type).unwrap(), + FieldFormat::Text, + )]); + assert_eq!(pg_schema[0].datatype(), &Type::JSON); + + let query_context = QueryContextBuilder::default() + .configuration_parameter(Default::default()) + .build() + .into(); + let encoder = DataRowEncoder::new(pg_schema.clone()); + let row_stream = RecordBatchRowStream::new( + query_context, + pg_schema.clone(), + schema, + stream::once(async { Ok(record_batch) }), + encoder, + ); + + let data_rows: Vec = futures::executor::block_on( + row_stream + .filter_map(|x: PgWireResult<_>| async move { x.ok() }) + .flat_map(stream::iter) + .collect(), + ); + assert_eq!(data_rows.len(), 3); + + // Decode text-format fields: [i32 BE length][bytes], -1 for NULL. + let decoded: Vec> = data_rows + .iter() + .map(|row| { + assert_eq!(row.field_count, 1); + let len = i32::from_be_bytes(row.data[0..4].try_into().unwrap()); + if len < 0 { + return None; + } + Some(String::from_utf8(row.data[4..4 + len as usize].to_vec()).unwrap()) + }) + .collect(); + + // A NULL list row stays NULL. + assert_eq!(decoded[2], None); + + // A list of structs is a JSON array; null elements stay null. + let json: serde_json::Value = serde_json::from_str(decoded[0].as_deref().unwrap()).unwrap(); + assert_eq!( + json, + serde_json::json!([ + {"a": 1, "b": 1.5}, + null, + {"a": 3, "b": 3.5}, + ]) + ); + + // An empty list is an empty JSON array. + let json: serde_json::Value = serde_json::from_str(decoded[1].as_deref().unwrap()).unwrap(); + assert_eq!(json, serde_json::json!([])); + } + + #[test] + fn test_encode_struct_value_as_json() { + use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount}; + use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram}; + use api::helper::pb_value_to_value_ref; + use api::v1::Value as PbValue; + use common_query::native_histogram::{ + encode_native_histogram, native_histogram_column_schema, native_histogram_value_type, + }; + use datatypes::vectors::Helper; + use pgwire::messages::data::DataRow; + + fn histogram_value(histogram: &Histogram) -> Value { + let value_data = encode_native_histogram(histogram).unwrap(); + let pb_value = PbValue { + value_data: Some(value_data), + }; + let column_schema = native_histogram_column_schema().unwrap(); + Value::from(pb_value_to_value_ref( + &pb_value, + column_schema.datatype_extension.as_ref(), + )) + } + + // One integer-count histogram and one float-count histogram, mirroring + // what prometheus remote-write v2 stores. The expected JSON matches the + // HTTP query output for the same payloads. + let int_histogram = Histogram { + schema: 1, + sum: 10.0, + zero_threshold: 0.001, + reset_hint: 2, + start_timestamp: 1500, + positive_spans: vec![BucketSpan { + offset: 0, + length: 3, + }], + positive_deltas: vec![1, 2, -1], + negative_spans: vec![BucketSpan { + offset: -2, + length: 1, + }], + negative_deltas: vec![1], + count: Some(Count::CountInt(8)), + zero_count: Some(ZeroCount::ZeroCountInt(1)), + ..Default::default() + }; + let float_histogram = Histogram { + schema: 2, + sum: 20.0, + zero_threshold: 0.002, + reset_hint: 3, + start_timestamp: 2500, + positive_spans: vec![BucketSpan { + offset: 3, + length: 2, + }], + positive_counts: vec![2.0, 3.5], + count: Some(Count::CountFloat(6.0)), + zero_count: Some(ZeroCount::ZeroCountFloat(0.5)), + ..Default::default() + }; + + let rows = vec![ + histogram_value(&int_histogram), + histogram_value(&float_histogram), + Value::Null, + ]; + let vector = + Helper::try_from_row_into_vector(&rows, native_histogram_value_type()).unwrap(); + + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "greptime_native_histogram", + native_histogram_value_type().clone(), + true, + )])); + let record_batch = RecordBatch::new(schema.clone(), vec![vector]).unwrap(); + + let pg_schema = Arc::new(vec![FieldInfo::new( + "greptime_native_histogram".into(), + None, + None, + Type::JSON, + FieldFormat::Text, + )]); + + let query_context = QueryContextBuilder::default() + .configuration_parameter(Default::default()) + .build() + .into(); + let encoder = DataRowEncoder::new(pg_schema.clone()); + let row_stream = RecordBatchRowStream::new( + query_context, + pg_schema.clone(), + schema, + stream::once(async { Ok(record_batch) }), + encoder, + ); + + let data_rows: Vec = futures::executor::block_on( + row_stream + .filter_map(|x: PgWireResult<_>| async move { x.ok() }) + .flat_map(stream::iter) + .collect(), + ); + assert_eq!(data_rows.len(), 3); + + // Decode text-format fields: [i32 BE length][bytes], -1 for NULL. + let decoded: Vec> = data_rows + .iter() + .map(|row| { + assert_eq!(row.field_count, 1); + let len = i32::from_be_bytes(row.data[0..4].try_into().unwrap()); + if len < 0 { + return None; + } + Some(String::from_utf8(row.data[4..4 + len as usize].to_vec()).unwrap()) + }) + .collect(); + + // A NULL struct row stays NULL. + assert_eq!(decoded[2], None); + + let json: serde_json::Value = serde_json::from_str(decoded[0].as_deref().unwrap()).unwrap(); + assert_eq!( + json, + serde_json::json!({ + "count_f64": null, + "count_i64": 8, + "custom_values": [], + "negative_buckets_f64": [], + "negative_buckets_i64": [1], + "negative_span_lengths": [1], + "negative_span_offsets": [-2], + "positive_buckets_f64": [], + "positive_buckets_i64": [1, 3, 2], + "positive_span_lengths": [3], + "positive_span_offsets": [0], + "reset_hint": 2, + "schema": 1, + "start_timestamp": 1500, + "sum": 10.0, + "zero_count_f64": null, + "zero_count_i64": 1, + "zero_threshold": 0.001 + }) + ); + + let json: serde_json::Value = serde_json::from_str(decoded[1].as_deref().unwrap()).unwrap(); + assert_eq!( + json, + serde_json::json!({ + "count_f64": 6.0, + "count_i64": null, + "custom_values": [], + "negative_buckets_f64": [], + "negative_buckets_i64": [], + "negative_span_lengths": [], + "negative_span_offsets": [], + "positive_buckets_f64": [2.0, 3.5], + "positive_buckets_i64": [], + "positive_span_lengths": [2], + "positive_span_offsets": [3], + "reset_hint": 3, + "schema": 2, + "start_timestamp": 2500, + "sum": 20.0, + "zero_count_f64": 0.5, + "zero_count_i64": null, + "zero_threshold": 0.002 + }) + ); + } + #[test] fn test_encode_utf8_oid_alias_data() { let aliases = [ diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index fd02db2e5a1..37a1740ccd1 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -1264,6 +1264,61 @@ pub async fn setup_pg_server_with_user_provider( (instance.guard, Arc::new(pg_server)) } +/// Sets up a standalone instance with both a Prometheus remote-write HTTP app +/// (native histograms enabled) and a Postgres server attached, so native +/// histogram data written via remote-write can be queried over the Postgres +/// protocol. +pub async fn setup_pg_server_with_prom_native_histogram( + store_type: StorageType, + name: &str, +) -> (TestGuard, Router, Arc>) { + unsafe { + std::env::set_var("TZ", "UTC"); + } + + let instance = setup_standalone_instance(name, store_type).await; + + // Prometheus remote-write HTTP app with native histograms enabled. + let http_server = build_test_prom_server(instance.fe_instance().clone(), false, true) + .with_greptime_config_options(instance.opts.datanode_options().to_toml().unwrap()) + .build(); + let app = http_server.build(http_server.make_app()).unwrap(); + + // Postgres server on the same instance. + let runtime = RuntimeBuilder::default() + .worker_threads(2) + .thread_name("pg-runtime") + .build() + .unwrap(); + + let fe_pg_addr = format!("127.0.0.1:{}", ports::get_port()); + let opts = PostgresOptions { + addr: fe_pg_addr.clone(), + ..Default::default() + }; + let tls_server_config = Arc::new( + ReloadableTlsServerConfig::try_new(opts.tls.clone()) + .expect("Failed to load certificates and keys"), + ); + + let mut pg_server = Box::new(PostgresServer::new( + instance.fe_instance().clone(), + opts.tls.should_force_tls(), + tls_server_config, + 0, + runtime, + None, + None, + )); + + pg_server + .start(fe_pg_addr.parse::().unwrap()) + .await + .unwrap(); + + (instance.guard, app, Arc::new(pg_server)) +} + pub(crate) async fn prepare_another_catalog_and_schema(instance: &Instance) { let catalog_manager = instance .catalog_manager() diff --git a/tests-integration/tests/sql.rs b/tests-integration/tests/sql.rs index 6c58ab0246c..e3b66fc4447 100644 --- a/tests-integration/tests/sql.rs +++ b/tests-integration/tests/sql.rs @@ -31,7 +31,8 @@ use sqlx::{Connection, Executor, Row}; use tests_integration::test_util::{ StorageType, setup_mysql_server, setup_mysql_server_with_slow_query_threshold, setup_mysql_server_with_user_provider, setup_pg_server, - setup_pg_server_with_slow_query_threshold, setup_pg_server_with_user_provider, + setup_pg_server_with_prom_native_histogram, setup_pg_server_with_slow_query_threshold, + setup_pg_server_with_user_provider, }; use tokio_postgres::{Client, NoTls, SimpleQueryMessage}; @@ -94,6 +95,8 @@ macro_rules! sql_tests { test_mysql_federated_prepare_stmt, test_mysql_prepare_tql_and_show, test_postgres_extended_query_row_returning_statements, + test_postgres_native_histogram, + test_postgres_struct_types, test_declare_fetch_close_cursor, test_alter_update_on, ); @@ -879,6 +882,234 @@ pub async fn test_postgres_bytea(store_type: StorageType) { guard.remove_all().await; } +pub async fn test_postgres_native_histogram(store_type: StorageType) { + use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount}; + use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram}; + use axum::http::StatusCode; + use prost::Message; + use servers::http::test_helpers::TestClient; + use servers::prom_remote_write::v2::test_util as remote_write_v2; + use servers::prom_store; + + let (mut guard, app, fe_pg_server) = + setup_pg_server_with_prom_native_histogram(store_type, "test_postgres_native_histogram") + .await; + let addr = fe_pg_server.bind_addr().unwrap().to_string(); + let http_client = TestClient::new(app).await; + + // Ingest one integer-count and one float-count native histogram via + // prometheus remote-write v2. + let write_request = remote_write_v2::request_with_labels_and_histograms( + vec![ + (prom_store::METRIC_NAME_LABEL, "pg_native_histogram_seconds"), + ("job", "api"), + ("instance", "localhost:9090"), + ], + vec![ + Histogram { + count: Some(Count::CountInt(8)), + sum: 10.0, + schema: 1, + zero_threshold: 0.001, + zero_count: Some(ZeroCount::ZeroCountInt(1)), + negative_spans: vec![BucketSpan { + offset: -2, + length: 1, + }], + negative_deltas: vec![1], + positive_spans: vec![BucketSpan { + offset: 0, + length: 3, + }], + positive_deltas: vec![1, 2, -1], + reset_hint: 2, + timestamp: 3000, + start_timestamp: 1500, + ..Default::default() + }, + Histogram { + count: Some(Count::CountFloat(6.0)), + sum: 20.0, + schema: 2, + zero_threshold: 0.002, + zero_count: Some(ZeroCount::ZeroCountFloat(0.5)), + positive_spans: vec![BucketSpan { + offset: 3, + length: 2, + }], + positive_counts: vec![2.0, 3.5], + reset_hint: 3, + timestamp: 4000, + start_timestamp: 2500, + ..Default::default() + }, + ], + ); + let compressed = prom_store::snappy_compress(&write_request.encode_to_vec()).unwrap(); + let res = http_client + .post("/v1/prometheus/write") + .header("Content-Encoding", "snappy") + .header( + "Content-Type", + "application/x-protobuf;proto=io.prometheus.write.v2.Request", + ) + .body(compressed) + .send() + .await; + assert_eq!(res.status(), StatusCode::NO_CONTENT); + + let (client, connection) = tokio_postgres::connect(&format!("postgres://{addr}/public"), NoTls) + .await + .unwrap(); + let (tx, rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + connection.await.unwrap(); + tx.send(()).unwrap(); + }); + + let rows = client + .simple_query( + "select greptime_native_histogram from pg_native_histogram_seconds order by greptime_timestamp", + ) + .await + .unwrap(); + let jsons: Vec<&str> = rows + .iter() + .filter_map(|message| match message { + SimpleQueryMessage::Row(row) => row.get(0), + _ => None, + }) + .collect(); + assert_eq!(jsons.len(), 2); + + // The struct column is encoded as a JSON object, matching the HTTP query + // output for the same histograms. Before the fix every row was encoded as + // an empty struct (`{}`). + let histogram: serde_json::Value = serde_json::from_str(jsons[0]).unwrap(); + assert_eq!( + histogram, + serde_json::json!({ + "count_f64": null, + "count_i64": 8, + "custom_values": [], + "negative_buckets_f64": [], + "negative_buckets_i64": [1], + "negative_span_lengths": [1], + "negative_span_offsets": [-2], + "positive_buckets_f64": [], + "positive_buckets_i64": [1, 3, 2], + "positive_span_lengths": [3], + "positive_span_offsets": [0], + "reset_hint": 2, + "schema": 1, + "start_timestamp": 1500, + "sum": 10.0, + "zero_count_f64": null, + "zero_count_i64": 1, + "zero_threshold": 0.001 + }) + ); + + let histogram: serde_json::Value = serde_json::from_str(jsons[1]).unwrap(); + assert_eq!( + histogram, + serde_json::json!({ + "count_f64": 6.0, + "count_i64": null, + "custom_values": [], + "negative_buckets_f64": [], + "negative_buckets_i64": [], + "negative_span_lengths": [], + "negative_span_offsets": [], + "positive_buckets_f64": [2.0, 3.5], + "positive_buckets_i64": [], + "positive_span_lengths": [2], + "positive_span_offsets": [3], + "reset_hint": 3, + "schema": 2, + "start_timestamp": 2500, + "sum": 20.0, + "zero_count_f64": 0.5, + "zero_count_i64": null, + "zero_threshold": 0.002 + }) + ); + + drop(client); + rx.await.unwrap(); + + let _ = fe_pg_server.shutdown().await; + guard.remove_all().await; +} + +pub async fn test_postgres_struct_types(store_type: StorageType) { + let (mut guard, fe_pg_server) = setup_pg_server(store_type, "test_postgres_struct_types").await; + let addr = fe_pg_server.bind_addr().unwrap().to_string(); + + let (client, connection) = tokio_postgres::connect(&format!("postgres://{addr}/public"), NoTls) + .await + .unwrap(); + let (tx, rx) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + connection.await.unwrap(); + tx.send(()).unwrap(); + }); + + async fn query_one(client: &Client, sql: &str) -> String { + let messages = client.simple_query(sql).await.unwrap(); + messages + .iter() + .filter_map(|message| match message { + SimpleQueryMessage::Row(row) => row.get(0).map(|v| v.to_string()), + _ => None, + }) + .next() + .unwrap() + } + + // Struct and list-of-struct columns are serialized as JSON. View-typed + // struct fields must not fail row extraction, and a null struct inside a + // list stays null instead of becoming a struct of null fields. + let row = query_one(&client, "SELECT struct(arrow_cast('abc', 'Utf8View'))").await; + assert_eq!(row, "{\"c0\":\"abc\"}"); + + // '0102' casts to its UTF-8 bytes when interpreted as BinaryView. + let row = query_one(&client, "SELECT struct(arrow_cast('0102', 'BinaryView'))").await; + assert_eq!(row, "{\"c0\":[48,49,48,50]}"); + + let row = query_one(&client, "SELECT struct([struct(1), NULL])").await; + assert_eq!(row, "{\"c0\":[{\"c0\":1},null]}"); + + // Top-level control: the same null element in a bare list stays null too. + let row = query_one(&client, "SELECT [struct(1), NULL]").await; + assert_eq!(row, "[{\"c0\":1},null]"); + + // Unsupported arrow field types surface as query errors instead of + // dropping the connection. + let error = client + .simple_query("SELECT struct(arrow_cast('1', 'Decimal256(38, 10)'))") + .await + .unwrap_err(); + let message = match error.as_db_error() { + Some(db_error) => db_error.message().to_string(), + None => error.to_string(), + }; + assert!( + message.contains("Unsupported arrow data type"), + "unexpected error message: {message}" + ); + + // The connection stays usable after the error. + let row = query_one(&client, "SELECT struct(arrow_cast('abc', 'Utf8View'))").await; + assert_eq!(row, "{\"c0\":\"abc\"}"); + + drop(client); + rx.await.unwrap(); + + let _ = fe_pg_server.shutdown().await; + guard.remove_all().await; +} + pub async fn test_postgres_slow_query(store_type: StorageType) { let (mut guard, fe_pg_server) = setup_pg_server_with_slow_query_threshold( store_type,