You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "jadireddi (via GitHub)" <gi...@apache.org> on 2025/06/10 22:18:44 UTC
[PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
jadireddi opened a new pull request, #1863:
URL: https://github.com/apache/iggy/pull/1863
…time.
# **PR Submission Guidelines**
**Please remove this entire section before submitting your PR.**
**This section is only for your reference.**
🙌 **Thank you for contributing to `iggy`!**
To help us incorporate your changes efficiently, please adhere to the following guidelines:
## General Coding Remarks
- **Code Formatting**: Run `cargo fmt` to ensure your code adheres to the project's style.
- **Code Linting**: Run `cargo clippy --all-targets --all-features -- -D warnings` to make sure your code is lint-free.
- **Unit Testing**: Write or update unit tests to cover your changes.
- **Integration Testing**: Write or update integration tests to cover your changes.
- **Project Structure**: Follow the `iggy` project's structure and coding style.
- **Build Integrity**: Ensure your code compiles and runs error-free.
- **Check unused dependencies**: Run `cargo machete` to make sure no unused dependencies made their way into your changeset.
- **Sort dependencies**: Run `cargo sort --workspace` so that the content of the toml files stay ordered.
## Commit Message Rules
- **Description**: Provide a concise description of the changes.
- **Style**: Use an imperative style in the subject line (e.g., "Fix bug" rather than "Fixed bug" or "Fixes bug").
- **Brevity**: Keep the subject line under 80 characters.
- **Rationale**: Explain the 'why' and 'what' of your changes in the summary.
- **Details**: Use the body to elaborate on the 'how' of your changes.
- **Context**: Include 'before' and 'after' scenarios if applicable.
- **References**: Link any relevant issues or PRs in the message body.
**Remember:** Your contribution is essential to the success of `iggy`. Please ensure that your PR conforms to these guidelines for a swift and smooth integration process.
Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "jadireddi (via GitHub)" <gi...@apache.org>.
jadireddi commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149407702
##########
core/connectors/sdk/src/transforms/mod.rs:
##########
@@ -16,25 +16,69 @@
* under the License.
*/
+mod add_fields;
+mod delete_fields;
+mod filter_fields;
+pub mod json;
+mod update_fields;
+use crate::{DecodedMessage, Error, TopicMetadata};
+pub use add_fields::{AddFields, AddFieldsConfig, Field as AddField};
+use chrono::Utc;
+pub use delete_fields::{DeleteFields, DeleteFieldsConfig};
+pub use filter_fields::{
+ FilterFields, FilterFieldsConfig, FilterPattern, KeyPattern as FilterKeyPattern,
+ ValuePattern as FilterValuePattern,
+};
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
use std::sync::Arc;
-
-use add_fields::AddFields;
-use delete_fields::DeleteFields;
-use serde::{Deserialize, Serialize, de::DeserializeOwned};
use strum_macros::{Display, IntoStaticStr};
-use tracing::error;
+pub use update_fields::{Field as UpdateField, UpdateCondition, UpdateFields, UpdateFieldsConfig};
-use crate::{DecodedMessage, Error, TopicMetadata};
+/// The value of a field, either static or computed at runtime
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FieldValue {
+ Static(OwnedValue),
+ Computed(ComputedValue),
+}
+
+/// Types of computed values that can be generated at runtime
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Display, IntoStaticStr)]
+#[serde(rename_all = "snake_case")]
+pub enum ComputedValue {
+ #[strum(to_string = "date_time")]
+ DateTime,
+ #[strum(to_string = "timestamp_nanos")]
+ TimestampNanos,
+ #[strum(to_string = "timestamp_micros")]
+ TimestampMicros,
+ #[strum(to_string = "timestamp_millis")]
+ TimestampMillis,
+ #[strum(to_string = "timestamp_seconds")]
+ TimestampSeconds,
+ #[strum(to_string = "uuid_v4")]
+ UuidV4,
+ #[strum(to_string = "uuid_v7")]
+ UuidV7,
+}
-pub mod add_fields;
-pub mod delete_fields;
+/// Computes a value based on the specified computed value type
+pub fn compute_value(kind: &ComputedValue) -> OwnedValue {
Review Comment:
Thank you for the review. Fixed now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149321238
##########
core/connectors/sdk/src/transforms/mod.rs:
##########
@@ -16,25 +16,69 @@
* under the License.
*/
+mod add_fields;
+mod delete_fields;
+mod filter_fields;
+pub mod json;
+mod update_fields;
+use crate::{DecodedMessage, Error, TopicMetadata};
+pub use add_fields::{AddFields, AddFieldsConfig, Field as AddField};
+use chrono::Utc;
+pub use delete_fields::{DeleteFields, DeleteFieldsConfig};
+pub use filter_fields::{
+ FilterFields, FilterFieldsConfig, FilterPattern, KeyPattern as FilterKeyPattern,
+ ValuePattern as FilterValuePattern,
+};
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
use std::sync::Arc;
-
-use add_fields::AddFields;
-use delete_fields::DeleteFields;
-use serde::{Deserialize, Serialize, de::DeserializeOwned};
use strum_macros::{Display, IntoStaticStr};
-use tracing::error;
+pub use update_fields::{Field as UpdateField, UpdateCondition, UpdateFields, UpdateFieldsConfig};
-use crate::{DecodedMessage, Error, TopicMetadata};
+/// The value of a field, either static or computed at runtime
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FieldValue {
+ Static(OwnedValue),
+ Computed(ComputedValue),
+}
+
+/// Types of computed values that can be generated at runtime
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Display, IntoStaticStr)]
+#[serde(rename_all = "snake_case")]
+pub enum ComputedValue {
+ #[strum(to_string = "date_time")]
+ DateTime,
+ #[strum(to_string = "timestamp_nanos")]
+ TimestampNanos,
+ #[strum(to_string = "timestamp_micros")]
+ TimestampMicros,
+ #[strum(to_string = "timestamp_millis")]
+ TimestampMillis,
+ #[strum(to_string = "timestamp_seconds")]
+ TimestampSeconds,
+ #[strum(to_string = "uuid_v4")]
+ UuidV4,
+ #[strum(to_string = "uuid_v7")]
+ UuidV7,
+}
-pub mod add_fields;
-pub mod delete_fields;
+/// Computes a value based on the specified computed value type
+pub fn compute_value(kind: &ComputedValue) -> OwnedValue {
Review Comment:
Maybe this could be moved to `json/mod` as its JSON specific?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2140399295
##########
core/connectors/sdk/src/transforms/test_utils.rs:
##########
@@ -0,0 +1,104 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#[cfg(test)]
Review Comment:
please dont do it this way, make whole module under cfg test
##########
core/connectors/sdk/src/transforms/add_fields.rs:
##########
@@ -18,58 +18,32 @@
use serde::{Deserialize, Serialize};
use simd_json::OwnedValue;
-use strum_macros::{Display, IntoStaticStr};
use crate::{DecodedMessage, Error, Payload, TopicMetadata};
Review Comment:
please remove empty spaces between use items
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "jadireddi (via GitHub)" <gi...@apache.org>.
jadireddi commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149970650
##########
core/connectors/sdk/src/transforms/json/update_fields.rs:
##########
@@ -0,0 +1,224 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::compute_value;
+use crate::{
+ DecodedMessage, Error, Payload, TopicMetadata,
+ transforms::{FieldValue, UpdateFields, update_fields::UpdateCondition},
+};
+use simd_json::OwnedValue;
+
+impl UpdateFields {
+ pub(crate) fn transform_json(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload else {
+ return Ok(Some(message));
+ };
+
+ for f in &self.fields {
+ let present = map.contains_key(&f.key);
+ let pass = match &f.condition {
+ None | Some(UpdateCondition::Always) => true,
+ Some(UpdateCondition::KeyExists) => present,
+ Some(UpdateCondition::KeyNotExists) => !present,
+ };
+ if pass {
+ let val = match &f.value {
+ FieldValue::Static(v) => v.clone(),
+ FieldValue::Computed(c) => compute_value(c),
+ };
+ map.insert(f.key.clone(), val);
+ }
+ }
+
+ Ok(Some(message))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::transforms::json::test_utils::{
+ assert_is_uuid, create_raw_test_message, create_test_message, create_test_topic_metadata,
+ extract_json_object,
+ };
+ use crate::transforms::{
+ ComputedValue, FieldValue, Transform,
+ update_fields::{Field, UpdateCondition, UpdateFields, UpdateFieldsConfig},
+ };
+ use simd_json::OwnedValue;
+ use simd_json::prelude::TypedScalarValue;
+
+ #[test]
+ fn test_basic_update() {
+ let transform = UpdateFields::new(UpdateFieldsConfig {
+ fields: vec![Field {
+ key: "status".to_string(),
+ value: FieldValue::Static(OwnedValue::from("updated")),
+ condition: None,
+ }],
+ });
+ let msg = create_test_message(r#"{"id": 1, "status": "pending"}"#);
+ let result = transform
+ .transform_json(&create_test_topic_metadata(), msg)
+ .unwrap()
+ .unwrap();
+ let json_obj = extract_json_object(&result).unwrap();
+ assert_eq!(json_obj.len(), 2);
+ assert_eq!(json_obj["id"], 1);
+ assert_eq!(json_obj["status"], "updated");
+ }
+
+ #[test]
+ fn test_condition_key_exists() {
Review Comment:
Thank you. Fixed the test names as BDD naming convention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "jadireddi (via GitHub)" <gi...@apache.org>.
jadireddi commented on PR #1863:
URL: https://github.com/apache/iggy/pull/1863#issuecomment-2970314184
Thank you, @hubcio and @spetz, for the review. I’ll update the code according to your feedback.
I first split FilterField and UpdateField into independent transformations. Later I noticed that Add, Update, Delete, and Filter shared a lot of common code, so I moved that common code to common.rs. I will keep each Field.rs file as independent transformation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149118748
##########
core/connectors/sdk/src/transforms/update_fields.rs:
##########
@@ -0,0 +1,123 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::{Transform, TransformType};
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use strum_macros::{Display, IntoStaticStr};
+
+/// The value of a field, either static or computed at runtime
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FieldValue {
+ Static(OwnedValue),
+ Computed(ComputedValue),
+}
+
+/// Types of computed values that can be generated at runtime
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Display, IntoStaticStr)]
+#[serde(rename_all = "snake_case")]
+pub enum ComputedValue {
+ #[strum(to_string = "date_time")]
+ DateTime,
+ #[strum(to_string = "timestamp_nanos")]
+ TimestampNanos,
+ #[strum(to_string = "timestamp_micros")]
+ TimestampMicros,
+ #[strum(to_string = "timestamp_millis")]
+ TimestampMillis,
+ #[strum(to_string = "timestamp_seconds")]
+ TimestampSeconds,
+ #[strum(to_string = "uuid_v4")]
+ UuidV4,
+ #[strum(to_string = "uuid_v7")]
+ UuidV7,
+}
+
+/// Computes a value based on the specified computed value type
+pub fn compute_value(kind: &ComputedValue) -> OwnedValue {
Review Comment:
Could it be made as the shared function for both, add and update fields transforms?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149413700
##########
core/connectors/sdk/src/transforms/json/update_fields.rs:
##########
@@ -0,0 +1,224 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::compute_value;
+use crate::{
+ DecodedMessage, Error, Payload, TopicMetadata,
+ transforms::{FieldValue, UpdateFields, update_fields::UpdateCondition},
+};
+use simd_json::OwnedValue;
+
+impl UpdateFields {
+ pub(crate) fn transform_json(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload else {
+ return Ok(Some(message));
+ };
+
+ for f in &self.fields {
Review Comment:
Let's have this named e.g. `field` so that the variable is understandable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex merged PR #1863:
URL: https://github.com/apache/iggy/pull/1863
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #1863:
URL: https://github.com/apache/iggy/pull/1863#issuecomment-2970078782
@jadireddi thank you for the contribution, it looks good! As @hubcio mentioned, it'd make sense to create a `json` subdirectory (module) under existing `sdk/transforms`, where all these JSON related code could be placed. And then, in the existing transformations, we could implement the internal function to be invoked per specific payload format. Please check the following example below:
`sdk/transforms/add_fields.rs`
```rust
use super::{Transform, TransformType};
use crate::{DecodedMessage, Error, Payload, TopicMetadata};
use serde::{Deserialize, Serialize};
use strum_macros::{Display, IntoStaticStr};
#[derive(Debug, Serialize, Deserialize)]
pub struct AddFieldsConfig {
fields: Vec<Field>,
}
pub struct AddFields {
pub fields: Vec<Field>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct Field {
pub key: String,
pub value: FieldValue,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldValue {
Static(simd_json::OwnedValue),
Computed(ComputedValue),
}
#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Display, IntoStaticStr)]
#[serde(rename_all = "snake_case")]
pub enum ComputedValue {
#[strum(to_string = "date_time")]
DateTime,
#[strum(to_string = "timestamp_nanos")]
TimestampNanos,
#[strum(to_string = "timestamp_micros")]
TimestampMicros,
#[strum(to_string = "timestamp_millis")]
TimestampMillis,
#[strum(to_string = "timestamp_seconds")]
TimestampSeconds,
#[strum(to_string = "uuid_v4")]
UuidV4,
#[strum(to_string = "uuid_v7")]
UuidV7,
}
impl AddFields {
pub fn new(config: AddFieldsConfig) -> Self {
Self {
fields: config.fields,
}
}
}
impl Transform for AddFields {
fn r#type(&self) -> TransformType {
TransformType::AddFields
}
fn transform(
&self,
metadata: &TopicMetadata,
message: DecodedMessage,
) -> Result<Option<DecodedMessage>, Error> {
if self.fields.is_empty() {
return Ok(Some(message));
}
match &message.payload {
Payload::Json(_) => self.transform_json(metadata, message),
_ => return Ok(Some(message)),
}
}
}
```
And then the internal `json` module under `sdk/transforms/json` with the related transformations e.g.
`sdk/transforms/json/add_fields.rs`
```rust
use simd_json::OwnedValue;
use crate::{
DecodedMessage, Error, Payload, TopicMetadata,
transforms::add_fields::{AddFields, ComputedValue, FieldValue},
};
impl AddFields {
pub fn transform_json(
&self,
metadata: &TopicMetadata,
mut message: DecodedMessage,
) -> Result<Option<DecodedMessage>, Error> {
let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload else {
return Ok(Some(message));
};
for field in &self.fields {
match &field.value {
FieldValue::Static(value) => map.insert(field.key.clone(), value.clone()),
FieldValue::Computed(value) => match value {
ComputedValue::DateTime => {
map.insert(field.key.clone(), chrono::Utc::now().to_rfc3339().into())
}
ComputedValue::TimestampMillis => map.insert(
field.key.clone(),
chrono::Utc::now().timestamp_millis().into(),
),
ComputedValue::TimestampMicros => map.insert(
field.key.clone(),
chrono::Utc::now().timestamp_micros().into(),
),
ComputedValue::TimestampNanos => map.insert(
field.key.clone(),
chrono::Utc::now().timestamp_nanos_opt().into(),
),
ComputedValue::TimestampSeconds => {
map.insert(field.key.clone(), chrono::Utc::now().timestamp().into())
}
ComputedValue::UuidV4 => {
map.insert(field.key.clone(), uuid::Uuid::new_v4().to_string().into())
}
ComputedValue::UuidV7 => {
map.insert(field.key.clone(), uuid::Uuid::now_v7().to_string().into())
}
},
};
}
Ok(Some(message))
}
}
```
By following this approach we could simply add more transforms for the different formats and keep things separated from each other (by simply exposing a dedicated function to be called in the match at the top level).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149117217
##########
core/connectors/sdk/src/transforms/add_fields.rs:
##########
@@ -16,39 +16,31 @@
* under the License.
*/
+use super::{Transform, TransformType};
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
use serde::{Deserialize, Serialize};
use simd_json::OwnedValue;
use strum_macros::{Display, IntoStaticStr};
-use crate::{DecodedMessage, Error, Payload, TopicMetadata};
-
-use super::{Transform, TransformType};
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct AddFieldsConfig {
- fields: Vec<Field>,
-}
-
-pub struct AddFields {
- fields: Vec<Field>,
-}
-
+/// A field to be added to messages
#[derive(Debug, Serialize, Deserialize)]
-struct Field {
- key: String,
- value: FieldValue,
+pub struct Field {
+ pub key: String,
+ pub value: FieldValue,
}
+/// The value of a field, either static or computed at runtime
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
-enum FieldValue {
Review Comment:
The `FieldValue` and `ComputedValue` are the same for both, `add_fields` and `update_fields` - can we make this shared & reusable?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #1863:
URL: https://github.com/apache/iggy/pull/1863#issuecomment-2975227236
Thank you for the improvements! Let's make the mentioned parts reusable (if applicable), and then we could merge it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): Extend JSON field transformations in connectors run… [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1863:
URL: https://github.com/apache/iggy/pull/1863#discussion_r2149425080
##########
core/connectors/sdk/src/transforms/json/update_fields.rs:
##########
@@ -0,0 +1,224 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::compute_value;
+use crate::{
+ DecodedMessage, Error, Payload, TopicMetadata,
+ transforms::{FieldValue, UpdateFields, update_fields::UpdateCondition},
+};
+use simd_json::OwnedValue;
+
+impl UpdateFields {
+ pub(crate) fn transform_json(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload else {
+ return Ok(Some(message));
+ };
+
+ for f in &self.fields {
+ let present = map.contains_key(&f.key);
+ let pass = match &f.condition {
+ None | Some(UpdateCondition::Always) => true,
+ Some(UpdateCondition::KeyExists) => present,
+ Some(UpdateCondition::KeyNotExists) => !present,
+ };
+ if pass {
+ let val = match &f.value {
+ FieldValue::Static(v) => v.clone(),
+ FieldValue::Computed(c) => compute_value(c),
+ };
+ map.insert(f.key.clone(), val);
+ }
+ }
+
+ Ok(Some(message))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::transforms::json::test_utils::{
+ assert_is_uuid, create_raw_test_message, create_test_message, create_test_topic_metadata,
+ extract_json_object,
+ };
+ use crate::transforms::{
+ ComputedValue, FieldValue, Transform,
+ update_fields::{Field, UpdateCondition, UpdateFields, UpdateFieldsConfig},
+ };
+ use simd_json::OwnedValue;
+ use simd_json::prelude::TypedScalarValue;
+
+ #[test]
+ fn test_basic_update() {
+ let transform = UpdateFields::new(UpdateFieldsConfig {
+ fields: vec![Field {
+ key: "status".to_string(),
+ value: FieldValue::Static(OwnedValue::from("updated")),
+ condition: None,
+ }],
+ });
+ let msg = create_test_message(r#"{"id": 1, "status": "pending"}"#);
+ let result = transform
+ .transform_json(&create_test_topic_metadata(), msg)
+ .unwrap()
+ .unwrap();
+ let json_obj = extract_json_object(&result).unwrap();
+ assert_eq!(json_obj.len(), 2);
+ assert_eq!(json_obj["id"], 1);
+ assert_eq!(json_obj["status"], "updated");
+ }
+
+ #[test]
+ fn test_condition_key_exists() {
Review Comment:
In order to stick to the already existing test conventions, let's follow the sort of BDD naming convention e.g. `should_update_only_the_existing_keys` or so (this could be applied to all the tests). Feel free to use the naming which makes most sense for the particular test (given/when/then/should etc.).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org