* feat: metric batch 2s PoC
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* chore: max_concurrent_flushes
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* chore: work channel size
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
* feat(servers): add metrics and logs for pending rows batch flush
Add the `FLUSH_ELAPSED` histogram metric to track the duration of pending
rows batch flushes in the Prometheus store protocol handler. This provides
better observability into the performance and latency of the batcher.
Also update telemetry by:
- Recording elapsed time for both successful and failed flush operations.
- Adding an informational log upon successful flush including row count and duration.
- Including elapsed time in error logs when a flush fails.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat(servers): implement columnar batching for pending rows
Refactor PendingRowsBatcher to use columnar batching for the metrics
store. Incoming RowInsertRequests are now converted to RecordBatches,
partitioned, and flushed via BulkInsert requests to datanodes.
- Enhance MultiDimPartitionRule to handle scalar boolean predicates.
- Add metrics for tracking flush failures and dropped rows.
- Update dependencies to support columnar batching in servers.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat(servers): add backpressure for pending rows
Implement backpressure in PendingRowsBatcher by limiting in-flight
requests with a semaphore and making the submission wait for the flush
result. This ensures Prometheus write requests are throttled and only
return once the data has been successfully flushed to datanodes.
- Add max_inflight_requests to PromStoreOptions.
- Use oneshot channels to notify submitters of flush completion.
- Limit concurrent requests using a new inflight_semaphore.
- Update PendingRowsBatcher::submit to wait for the flush outcome.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat: add stage-level metrics for bulk ingestion
Introduce histograms to track the elapsed time of various stages in the
metric engine bulk insert path and the server's pending rows batcher.
This provides better observability into the performance bottlenecks
of the ingestion pipeline.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* - `src/metric-engine/src/engine/bulk_insert.rs`: Removed the fallback mechanism that converted record batches to rows when bulk inserts were unsupported, along with related helper functions and unused imports.
- `src/operator/src/insert.rs`: Removed an unused import (`common_time::TimeToLive::Instant`).
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat(servers): columnar Prom remote write
Optimize the Prometheus remote write path by allowing direct conversion
from decoded Prometheus samples to Arrow RecordBatches. This bypasses
intermediate row-based representations when `PendingRowsBatcher` is
active and no pipeline is used, improving ingestion efficiency.
- Implement `as_record_batch_groups` in `TablesBuilder` and `PromWriteRequest`.
- Add `submit_prom_record_batch_groups` to `PendingRowsBatcher`.
- Introduce `DecodedPromWriteRequest` in `prom_store`.
- Implement row-to-RecordBatch conversion logic in `prom_row_builder`.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* Revert "feat(servers): columnar Prom remote write"
This reverts commit efbb63c12a3e7fcec03858ea0351efd94fec8242.
* refactor(servers): improve row to RecordBatch conversion
- Use `snafu::ensure` for row validation in `rows_to_record_batch`.
- Add explicit type hint for `MutableVector` to improve clarity.
- Reorganize and clean up imports in `pending_rows_batcher.rs`.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* perf(servers): use arrow builders for row conversion
This commit optimizes the conversion from `api::v1::Rows` to `RecordBatch`
by using Arrow builders directly. This avoids the overhead of
`MutableVector` and `common_recordbatch`, leading to better performance
in the `pending_rows_batcher`.
Additionally, the `#[allow(dead_code)]` attribute is removed from
`modify_batch_sparse` in the metric engine as it is now utilized.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* perf(metric-engine): optimize batch modification
Optimize `modify_batch_sparse` by reusing buffers, using Arrow
builders, and employing fast-path encoding methods. This reduces
allocations and avoids redundant downcasting and serializer overhead.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat/metric-engine-support-bulk:
**Add Environment Variable for Batch Sync Control**
- `pending_rows_batcher.rs`: Introduced an environment variable `PENDING_ROWS_BATCH_SYNC` to control the synchronization behavior of batch processing. If set to true, the function will wait for the flush result; otherwise, it will return immediatel
with the total rows count.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* wip
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* chore: update and fix clippy
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* fix: failing test
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* picking-pending-rows-batcher:
### Commit Message
Remove Unused Code and Simplify Error Handling
- **`src/error.rs`**: Removed the `BatcherQueueFull` error variant and its associated logic, simplifying the error handling by removing unused code.
- **`src/http/prom_store.rs`**: Eliminated the `try_decompress` function, streamlining the decompression logic by directly using `snappy_decompress` in `decode_remote_read_request`.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* chore: parse PENDING_ROWS_BATCH_SYNC once
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* chore: revert unrelated changes
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* **Refactor Prometheus Write Handling**
- **`prom_store.rs`**: Introduced `pre_write` method in `PromStoreProtocolHandler` to handle pre-write checks for Prometheus remote write requests. Updated `write` method to utilize `pre_write`.
- **`server.rs`**: Modified `PendingRowsBatcher` initialization to conditionally create a batcher based on `with_metric_engine` flag.
- **`http/prom_store.rs`**: Integrated `pre_write` checks before submitting requests to `PendingRowsBatcher`.
- **`query_handler.rs`**: Added `pre_write` method to `PromStoreProtocolHandler` trait for pre-write operations.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* picking-pending-rows-batcher:
- **Fix Label Typo**: Corrected a typo in the label value from `"flush_wn ite_region"` to `"flush_write_region"` in `pending_rows_batcher.rs`.
- **Refactor Array Building Logic**: Introduced a macro `build_array!` to streamline the construction of `ArrayRef` for different data types, reducing code duplication in `pending_rows_batcher.rs`.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* format toml
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* picking-pending-rows-batcher:
### Update PromStore and PendingRowsBatcher Configuration
- **`prom_store.rs`**: Set `pending_rows_flush_interval` to `Duration::ZERO` to disable automatic flushing.
- **`pending_rows_batcher.rs`**: Enhance validation to disable the batcher when `flush_interval` is zero or configuration values like `max_batch_rows`, `max_concurrent_flushes`, `worker_channel_capacity`, or `max_inflight_requests` are zero, preventing potential panics or deadlocks.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* picking-pending-rows-batcher:
### Update `pending_rows_flush_interval` to Zero
- **Files Modified**:
- `src/frontend/src/service_config/prom_store.rs`
- `tests-integration/tests/http.rs`
- **Key Changes**:
- Updated `pending_rows_flush_interval` from `Duration::from_secs(2)` to `Duration::ZERO` in `prom_store.rs`.
- Changed `pending_rows_flush_interval` configuration from `"2s"` to `"0s"` in `http.rs`.
These changes set the flush interval to zero, potentially affecting how frequently pending rows are flushed.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* picking-pending-rows-batcher:
**Add Worker Management Enhancements**
- **`metrics.rs`**: Introduced `PENDING_WORKERS` gauge to track active pending rows batch workers.
- **`pending_rows_batcher.rs`**:
- Added worker idle timeout logic with `WORKER_IDLE_TIMEOUT_MULTIPLIER`.
- Implemented worker management functions: `spawn_worker`, `remove_worker_if_same_channel`, and `should_close_worker_on_idle_timeout`.
- Enhanced worker lifecycle management to handle idle workers and ensure proper cleanup.
- **Tests**: Added unit tests for worker removal and idle timeout logic.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* fix: clippy
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
---------
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
Co-authored-by: jeremyhi <fengjiachun@gmail.com>
chore: remove GrpcQueryHandler::put_record_batch, we should use GrpcQueryHandler::handle_put_record_batch_stream instead
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat: initial function rewriter for json_get
* feat: make sure rewrite rule is applied
* feat: keep analyzer's default rules
* feat: implement rewriter for arrow_cast
* test: add unit test for tht rewriter
* chore: format
* refactor: extract some more functions
* Apply suggestion from @waynexia
Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
---------
Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
* fix/flight-stuck-on-first-message:
**Refactor GRPC Stream Handling and Table Resolution**
- **`grpc.rs`**: Refactored the `GrpcQueryHandler` to resolve table references and check permissions only once per stream, improving efficiency. Introduced a mechanism to handle table resolution and permission checks after receiving the first `RecordBatch`.
- **`flight.rs`**: Enhanced `PutRecordBatchRequestStream` to manage stream states (`Init` and `Ready`) for better handling of schema and table name extraction. Improved error handling and logging for unexpected flight messages.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* chore: add some doc
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
---------
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
refactor: decode FlightData early in put_record_batch pipeline
- Move FlightDecoder usage from Inserter up to PutRecordBatchRequestStream,
passing decoded RecordBatch and schema bytes instead of raw FlightData.
- Eliminate redundant per-request decoding/encoding in Inserter; encode
once and reuse for all region requests.
- Streamline GrpcQueryHandler trait and implementations to accept
PutRecordBatchRequest containing pre-decoded data.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
feat: stream-based bulk insert with per-batch responses
- Introduce handle_put_record_batch_stream() to process Flight DoPut streams
- Resolve table & permissions once, yield (request_id, AffectedRows) per batch
- Replace loop-over-request with async-stream in frontend & server
- Make PutRecordBatchRequestStream public for cross-crate usage
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
fix: propagate request_id with errors in bulk insert stream
Changes the bulk-insert stream item type from
Result<(i64, AffectedRows), E> to (i64, Result<AffectedRows, E>)
so every emitted tuple carries the request_id even on failure,
letting callers correlate errors with the originating request.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
refactor: unify DoPut response stream to return DoPutResponse
Replace the tuple (i64, Result<AffectedRows>) with Result<DoPutResponse>
throughout the gRPC bulk-insert path so the handler, adapter and server
all speak the same type.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
feat: add elapsed_secs to DoPutResponse for bulk-insert timing
- DoPutResponse now carries elapsed_secs field
- Frontend measures and attaches insert duration
- Server observes GRPC_BULK_INSERT_ELAPSED metric from response
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
refactor: unify Bytes import in flight module
- Replace `bytes::Bytes` with `Bytes` alias for consistency
- Remove redundant `ProstBytes` alias
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
fix: terminate gRPC stream on error and optimize FlightData handling
- Stop retrying on stream errors in gRPC handler
- Replace Vec1 indexing with into_iter().next() for FlightData
- Remove redundant clones in bulk_insert and flight modules
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
Improve permission check placement in `grpc.rs`
- Moved the permission check for `BulkInsert` to occur before resolving the table reference in `GrpcQueryHandler` implementation.
- Ensures permission validation is performed earlier in the process, potentially avoiding unnecessary operations if permission is denied.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
**Refactor Bulk Insert Handling in gRPC**
- **`grpc.rs`**:
- Switched from `async_stream::stream` to `async_stream::try_stream` for error handling.
- Removed `body_size` parameter and added `flight_data` to `handle_bulk_insert`.
- Simplified error handling and permission checks in `GrpcQueryHandler`.
- **`bulk_insert.rs`**:
- Added `raw_flight_data` parameter to `handle_bulk_insert`.
- Calculated `body_size` from `raw_flight_data` and removed redundant encoding logic.
- **`flight.rs`**:
- Replaced `body_size` with `flight_data` in `PutRecordBatchRequest`.
- Updated memory usage calculation to include `flight_data` components.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/bulk-insert-service:
perf(bulk_insert): encode record batch once per datanode
Move FlightData encoding outside the per-region loop so the same
encoded bytes are reused when mask.select_all(), eliminating redundant
serialisation work.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
---------
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/allow-custom-flight-service:
### Add Custom Flight Handler Support
- **`server.rs`**:
- Introduced a new field `flight_handler` in the `Services` struct to allow optional custom flight handler configuration.
- Added a method `with_flight_handler` to set the custom flight handler.
- Modified `build_grpc_server` to use the custom flight handler if provided, defaulting to `GreptimeRequestHandler` otherwise.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* refactor/allow-custom-flight-service:
### Make structs and enums public in `flight.rs`
- Changed visibility of `PutRecordBatchRequest` and `PutRecordBatchRequestStream` structs to public.
- Made `PutRecordBatchRequestStreamState` enum public.
- Updated fields within `PutRecordBatchRequest` and `PutRecordBatchRequestStream` to be public.
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
---------
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
* feat: struct value
Signed-off-by: Ning Sun <sunning@greptime.com>
* feat: update for proto module
* feat: wip struct type
* feat: implement more vector operations
* feat: make datatype and api
* feat: reoslve some compilation issues
* feat: resolve all compilation issues
* chore: format update
* test: resolve tests
* test: test and refactor value-to-pb
* feat: add more tests and fix for value types
* chore: remove dbg
* feat: test and fix iterator
* fix: resolve struct_type issue
* feat: pgwire 0.33 update
* refactor: use vec for struct items
* feat: conversion from json to value
* feat: add decode function
* fix: lint issue
* feat: update how we encode raw data
* feat: add convertion to fully strcutured StructValue
* refactor: take owned value in all encode/decode functions
* feat: add pg serialization of structvalue
* chore: toml format
* refactor: adopt new and try_new from struct value
* chore: cleanup residual issues
* docs: docs up
* fix lint issue
* Apply suggestion from @MichaelScofield
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
* Apply suggestion from @MichaelScofield
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
* Apply suggestion from @MichaelScofield
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
* Apply suggestion from @MichaelScofield
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
* chore: address review comment especially collection capacity
* refactor: remove unneeded processed keys collection
* feat: Value::Json type
* chore: add some work in progress changes
* feat: adopt new json type
* refactor: limit scope json conversion functions
* fix: self review update
* test: provide tests for value::json
* test: add tests for api/helper
* switch proto to main branch
* fix: implement is_null for ValueRef::Json
---------
Signed-off-by: Ning Sun <sunning@greptime.com>
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>
* feat: struct value
Signed-off-by: Ning Sun <sunning@greptime.com>
* feat: update for proto module
* feat: wip struct type
* feat: implement more vector operations
* feat: make datatype and api
* feat: reoslve some compilation issues
* feat: resolve all compilation issues
* chore: format update
* test: resolve tests
* test: test and refactor value-to-pb
* feat: add more tests and fix for value types
* chore: remove dbg
* feat: test and fix iterator
* fix: resolve struct_type issue
* refactor: use vec for struct items
* chore: update proto to main branch
* refactor: address some of review issues
* refactor: update for further review
* Add validation on new methods
* feat: update struct/list json serialization
* refactor: reimplement get in struct_vector
* refactor: struct vector functions
* refactor: fix lint issue
* refactor: address review comments
---------
Signed-off-by: Ning Sun <sunning@greptime.com>
* feat: align influxdb line timestamp with table time index
Signed-off-by: luofucong <luofc@foxmail.com>
* fix ci
Signed-off-by: luofucong <luofc@foxmail.com>
---------
Signed-off-by: luofucong <luofc@foxmail.com>
