rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-04 09:57:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into chore/stable-toolchain-1.96

Browse Source
This commit is contained in:
Ning Sun
2026-09-23 10:56:09 +08:00
committed by GitHub
parent a9431c17ad 0f118bc5ba
commit ceaec52d94
34 changed files with 1516 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/actions/build-linux-artifacts/action.yml
+23
View File
@@ -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 \
.github/workflows/dev-build.yml
+8
View File
@@ -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
.github/workflows/nightly-build.yml
+8
View File
@@ -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
.github/workflows/nightly-ci.yml
+52 -1
View File
@@ -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:
.github/workflows/release.yml
+8
View File
@@ -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
.github/workflows/rust.yml
+16 -18
View File
@@ -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
Makefile
+1
View File
@@ -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}
src/common/event-recorder/src/recorder.rs
+22 -8
View File
@@ -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<dyn EventHandler>) -> 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<dyn EventHandler>,
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(
src/datatypes/src/data_type.rs
+3 -1
View File
@@ -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(_)
src/datatypes/src/types/struct_type.rs
+19
View File
@@ -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<Self> {
let struct_fields = fields
.iter()
.map(|field| {
Ok(StructField::new(
field.name().clone(),
ConcreteDataType::try_from(field.data_type())?,
field.is_nullable(),
))
})
.collect::<Result<Vec<_>>>()?;
Ok(StructType {
fields: Arc::new(struct_fields),
})
}
pub fn fields(&self) -> Arc<Vec<StructField>> {
self.fields.clone()
}
src/datatypes/src/value.rs
+94 -5
View File
@@ -1152,12 +1152,13 @@ impl TryFrom<ScalarValue> 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<ScalarValue> 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<ScalarValue> for Value {
| ScalarValue::LargeListView(_)
| ScalarValue::Union(_, _, _)
| ScalarValue::Float16(_)
| ScalarValue::Utf8View(_)
| ScalarValue::BinaryView(_)
| ScalarValue::Map(_)
| ScalarValue::Date64(_)
| ScalarValue::RunEndEncoded(_, _, _) => {
@@ -1259,6 +1266,21 @@ impl TryFrom<ScalarValue> 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<Value> {
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<ValueRef<'_>> 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]
src/frontend/src/instance/builder.rs
+1
View File
@@ -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);
src/meta-srv/src/metasrv/builder.rs
+1
View File
@@ -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));
src/object-store/src/test_util.rs
-29
View File
@@ -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<String>,
}
/// Returns s3 test config, return None if not found.
pub fn s3_test_config() -> Option<TestS3Config> {
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
}
src/object-store/tests/object_store_test.rs
+9 -1
View File
@@ -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);
src/servers/src/postgres/types.rs
+308 -7
View File
@@ -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<S: Encoder>(
_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<S: Encoder>(
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<S, B>
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::<i32>();
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<Type> {
&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<DataRow> = 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<Option<String>> = 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<DataRow> = 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<Option<String>> = 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 = [
tests-integration/src/cluster.rs
+14
View File
@@ -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<Arc<TempDir>>,
meta_selector: Option<SelectorRef>,
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()
};
tests-integration/src/test_util.rs
+110 -24
View File
@@ -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_path>/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<TempFolder>,
/// Local read cache directory. Only held so it is removed when the guard drops.
#[allow(dead_code)]
local_cache: Option<TempDir>,
}
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);
}
@@ -1233,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<Box<dyn Server>>) {
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::<SocketAddr>().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()
tests-integration/src/tests/instance_test.rs
+15 -183
View File
@@ -2195,47 +2195,6 @@ async fn test_delete(instance: Arc<dyn MockInstance>) {
check_output_stream(output, expect).await;
}
#[apply(both_instances_cases)]
async fn test_execute_copy_to_s3(instance: Arc<dyn MockInstance>) {
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, &copy_to_stmt).await.data;
assert!(matches!(output, OutputData::AffectedRows(2)));
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_from_s3(instance: Arc<dyn MockInstance>) {
common_telemetry::init_default_ut_logging();
@@ -2268,9 +2227,21 @@ async fn test_execute_copy_from_s3(instance: Arc<dyn MockInstance>) {
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, &copy_to_stmt).await.data;
@@ -2315,10 +2286,7 @@ async fn test_execute_copy_from_s3(instance: Arc<dyn MockInstance>) {
.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<dyn MockInstance>) {
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_to_oss(instance: Arc<dyn MockInstance>) {
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, &copy_to_stmt).await.data;
assert!(matches!(output, OutputData::AffectedRows(2)));
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_from_oss(instance: Arc<dyn MockInstance>) {
common_telemetry::init_default_ut_logging();
@@ -2490,54 +2414,6 @@ async fn test_execute_copy_from_oss(instance: Arc<dyn MockInstance>) {
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_to_gcs(instance: Arc<dyn MockInstance>) {
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, &copy_to_stmt).await.data;
assert!(matches!(output, OutputData::AffectedRows(2)));
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_from_gcs(instance: Arc<dyn MockInstance>) {
common_telemetry::init_default_ut_logging();
@@ -2646,50 +2522,6 @@ async fn test_execute_copy_from_gcs(instance: Arc<dyn MockInstance>) {
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_to_azblob(instance: Arc<dyn MockInstance>) {
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, &copy_to_stmt).await.data;
assert!(matches!(output, OutputData::AffectedRows(2)));
}
}
#[apply(both_instances_cases)]
async fn test_execute_copy_from_azblob(instance: Arc<dyn MockInstance>) {
common_telemetry::init_default_ut_logging();
tests-integration/src/tests/promql_test.rs
+39 -10
View File
@@ -644,28 +644,57 @@ async fn two_aggregators_combined_aggrs(instance: Arc<dyn MockInstance>) {
// 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<dyn MockInstance>) {
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::<Vec<_>>();
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::<StringArray>().unwrap();
let values = batches[0]
.column_by_name("stddev_pop(http_requests.value)")
.unwrap()
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let mut actual = (0..values.len())
.map(|i| (instances.value(i), values.value(i)))
.collect::<Vec<_>>();
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
tests-integration/tests/admin_function_event.rs
+2 -23
View File
@@ -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<Instance>) {
execute(
instance,
tests-integration/tests/database_ddl_event.rs
+5 -2
View File
@@ -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();
tests-integration/tests/export_logical_tables.rs
+7 -2
View File
@@ -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) {
tests-integration/tests/flow_ddl_event.rs
+4 -1
View File
@@ -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();
tests-integration/tests/gc_event.rs
+3
View File
@@ -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)
tests-integration/tests/grpc.rs
+126 -21
View File
@@ -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;
}
tests-integration/tests/http.rs
+361 -35
View File
@@ -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 <user:pass>` + 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::<Value>(&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::<Vec<_>>();
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::<i64>() {
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::<Vec<_>>()
.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())
tests-integration/tests/main.rs
+11
View File
@@ -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;
tests-integration/tests/reconciliation_event.rs
+4
View File
@@ -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;
tests-integration/tests/repartition_event.rs
+2
View File
@@ -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))
tests-integration/tests/sql.rs
+232 -1
View File
@@ -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,
tests-integration/tests/table_ddl_event.rs
+4 -1
View File
@@ -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 {
tests-integration/tests/view_ddl_event.rs
+2 -14
View File
@@ -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<frontend::instance::Instance>,
source_table: &str,
tests-integration/tests/wal_prune_event.rs
+2
View File
@@ -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());