Merge remote-tracking branch 'origin/main' into reflect

Author: Devdutt Shenoi

.github/workflows/build-push-edge-debug.yaml

@@ -45,5 +45,3 @@ jobs:
           push: true
           tags: parseable/parseable:edge-debug
           platforms: linux/amd64,linux/arm64
-          cache-from: type=gha
-          cache-to: type=gha,mode=max

.github/workflows/build-push-edge.yaml

@@ -45,5 +45,3 @@ jobs:
           push: true
           tags: parseable/parseable:edge
           platforms: linux/amd64,linux/arm64
-          cache-from: type=gha
-          cache-to: type=gha,mode=max

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "parseable"
-version = "1.7.4"
+version = "1.7.5"
 authors = ["Parseable Team <[email protected]>"]
 edition = "2021"
 rust-version = "1.83.0"
@@ -128,7 +128,6 @@ sha1_smol = { version = "1.0", features = ["std"] }
 static-files = "0.2"
 ureq = "2.12"
 url = "2.5"
-vergen = { version = "9.0", features = ["build", "cargo", "rustc", "si"] }
 vergen-gitcl = { version = "1.0", features = ["build", "cargo", "rustc", "si"] }
 zip = { version = "2.3", default-features = false, features = ["deflate"] }
 anyhow = "1.0"

Cargo.toml

@@ -14,30 +14,23 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 # build stage
-FROM  rust:1.84.0-bookworm AS builder
-
+FROM rust:1.84.0-bookworm AS builder
 
 LABEL org.opencontainers.image.title="Parseable"
 LABEL maintainer="Parseable Team <[email protected]>"
 LABEL org.opencontainers.image.vendor="Parseable Inc"
 LABEL org.opencontainers.image.licenses="AGPL-3.0"
 
 WORKDIR /parseable
-
-# Cache dependencies
-COPY Cargo.toml Cargo.lock build.rs ./
-RUN mkdir src && echo "fn main() {}" > src/main.rs && cargo build --release && rm -rf src
-
-# Build the actual binary
-COPY src ./src
+COPY . .
 RUN cargo build --release
 
 # final stage
 FROM gcr.io/distroless/cc-debian12:latest
 
 WORKDIR /parseable
 
-# Copy the static binary into the final image
+# Copy the static shell into base image.
 COPY --from=builder /parseable/target/release/parseable /usr/bin/parseable
 
 CMD ["/usr/bin/parseable"]

Dockerfile

@@ -270,6 +270,14 @@ pub struct Options {
     )]
     pub row_group_size: usize,
 
+    #[arg(
+        long,
+        env = "P_EXECUTION_BATCH_SIZE",
+        default_value = "20000",
+        help = "batch size for query execution"
+    )]
+    pub execution_batch_size: usize,
+
     #[arg(
         long = "compression-algo",
         env = "P_PARQUET_COMPRESSION_ALGO",

src/cli.rs

@@ -16,20 +16,20 @@
  *
  */
 
-use std::sync::Arc;
-
 use async_trait::async_trait;
 use futures_util::StreamExt;
 use rdkafka::consumer::{CommitMode, Consumer};
 use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::Arc;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, error};
 
 use crate::{
     connectors::common::processor::Processor,
     event::{
         format::{json, EventFormat, LogSourceEntry},
-        Event as ParseableEvent,
+        Event as ParseableEvent, USER_AGENT_KEY,
     },
     parseable::PARSEABLE,
     storage::StreamType,
@@ -76,6 +76,9 @@ impl ParseableSinkProcessor {
             }
         }
 
+        let mut p_custom_fields = HashMap::new();
+        p_custom_fields.insert(USER_AGENT_KEY.to_string(), "kafka".to_string());
+
         let p_event = json::Event::new(Value::Array(json_vec)).into_event(
             stream_name.to_string(),
             total_payload_size,
@@ -85,6 +88,7 @@ impl ParseableSinkProcessor {
             time_partition.as_ref(),
             schema_version,
             StreamType::UserDefined,
+            &p_custom_fields,
         )?;
 
         Ok(p_event)

src/connectors/kafka/processor.rs

@@ -149,6 +149,7 @@ impl EventFormat for Event {
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
         stream_type: StreamType,
+        p_custom_fields: &HashMap<String, String>,
     ) -> Result<super::Event, anyhow::Error> {
         let custom_partition_values = match custom_partitions.as_ref() {
             Some(custom_partition) => {
@@ -168,6 +169,7 @@ impl EventFormat for Event {
             static_schema_flag,
             time_partition,
             schema_version,
+            p_custom_fields,
         )?;
 
         Ok(super::Event {

src/event/format/json.rs

@@ -33,7 +33,7 @@ use serde_json::Value;
 use crate::{
     metadata::SchemaVersion,
     storage::StreamType,
-    utils::arrow::{get_field, get_timestamp_array, replace_columns},
+    utils::arrow::{add_parseable_fields, get_field},
 };
 
 use super::{Event, DEFAULT_TIMESTAMP_KEY};
@@ -145,9 +145,10 @@ pub trait EventFormat: Sized {
         static_schema_flag: bool,
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
+        p_custom_fields: &HashMap<String, String>,
     ) -> Result<(RecordBatch, bool), AnyError> {
         let p_timestamp = self.get_p_timestamp();
-        let (data, mut schema, is_first) = self.to_data(
+        let (data, schema, is_first) = self.to_data(
             storage_schema,
             time_partition,
             schema_version,
@@ -161,16 +162,6 @@ pub trait EventFormat: Sized {
             ));
         };
 
-        // add the p_timestamp field to the event schema to the 0th index
-        schema.insert(
-            0,
-            Arc::new(Field::new(
-                DEFAULT_TIMESTAMP_KEY,
-                DataType::Timestamp(TimeUnit::Millisecond, None),
-                true,
-            )),
-        );
-
         // prepare the record batch and new fields to be added
         let mut new_schema = Arc::new(Schema::new(schema));
         if !Self::is_schema_matching(new_schema.clone(), storage_schema, static_schema_flag) {
@@ -179,12 +170,8 @@ pub trait EventFormat: Sized {
         new_schema =
             update_field_type_in_schema(new_schema, None, time_partition, None, schema_version);
 
-        let mut rb = Self::decode(data, new_schema.clone())?;
-        rb = replace_columns(
-            rb.schema(),
-            &rb,
-            &[(0, Arc::new(get_timestamp_array(p_timestamp, rb.num_rows())))],
-        );
+        let rb = Self::decode(data, new_schema.clone())?;
+        let rb = add_parseable_fields(rb, p_timestamp, p_custom_fields)?;
 
         Ok((rb, is_first))
     }
@@ -222,6 +209,7 @@ pub trait EventFormat: Sized {
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
         stream_type: StreamType,
+        p_custom_fields: &HashMap<String, String>,
     ) -> Result<Event, AnyError>;
 }
 

src/event/format/mod.rs

@@ -35,6 +35,9 @@ use chrono::NaiveDateTime;
 use std::collections::HashMap;
 
 pub const DEFAULT_TIMESTAMP_KEY: &str = "p_timestamp";
+pub const USER_AGENT_KEY: &str = "p_user_agent";
+pub const SOURCE_IP_KEY: &str = "p_src_ip";
+pub const FORMAT_KEY: &str = "p_format";
 
 #[derive(Clone)]
 pub struct Event {

src/event/mod.rs

@@ -23,6 +23,7 @@ use actix_web::{
     middleware::Next,
 };
 use actix_web_httpauth::extractors::basic::BasicAuth;
+use http::header::USER_AGENT;
 use ulid::Ulid;
 
 use crate::{
@@ -85,7 +86,7 @@ pub async fn audit_log_middleware(
         )
         .with_user_agent(
             req.headers()
-                .get("User-Agent")
+                .get(USER_AGENT)
                 .and_then(|a| a.to_str().ok())
                 .unwrap_or_default(),
         )