DOC-996 Add schema_evolution.processors (#173)

Author: asimms41

modules/components/pages/inputs/redpanda_migrator.adoc

@@ -58,7 +58,7 @@ input:
     topics: [] # No default (required)
     regexp_topics: false
     rack_id: "" # No default (optional)
-    instance_id "" # No default (optional)
+    instance_id: "" # No default (optional)
     rebalance_timeout: 45s
     session_timeout: 1m
     heartbeat_interval: 3s

modules/components/pages/outputs/snowflake_streaming.adoc

@@ -38,19 +38,12 @@ output:
       CREATE TABLE IF NOT EXISTS mytable (amount NUMBER);
     schema_evolution:
       enabled: false # No default (required)
-      new_column_type_mapping: |-
-        root = match this.value.type() {
-          this == "string" => "STRING"
-          this == "bytes" => "BINARY"
-          this == "number" => "DOUBLE"
-          this == "bool" => "BOOLEAN"
-          this == "timestamp" => "TIMESTAMP"
-          _ => "VARIANT"
+      processors: [] # No default (optional)
     batching:
       count: 0
       byte_size: 0
-      period: ""
-      check: ""
+      period: "" # No default (optional)
+      check: "" # No default (optional)
     max_in_flight: 4
 ```
 
@@ -78,23 +71,15 @@ output:
       CREATE TABLE IF NOT EXISTS mytable (amount NUMBER);
     schema_evolution:
       enabled: false # No default (required)
-      new_column_type_mapping: |-
-        root = match this.value.type() {
-          this == "string" => "STRING"
-          this == "bytes" => "BINARY"
-          this == "number" => "DOUBLE"
-          this == "bool" => "BOOLEAN"
-          this == "timestamp" => "TIMESTAMP"
-          _ => "VARIANT"
-        }
+      processors: [] # No default (optional)
     build_options:
       parallelism: 1
       chunk_size: 50000
     batching:
       count: 0
       byte_size: 0
-      period: ""
-      check: ""
+      period: "" # No default (optional)
+      check: "" # No default (optional)
       processors: [] # No default (optional)
     max_in_flight: 4
     channel_prefix: channel-${HOST} # No default (optional)
@@ -327,28 +312,23 @@ Whether schema evolution is enabled. When set to `true`, the Snowflake table is
 
 *Type*: `bool`
 
-=== `schema_evolution.new_column_type_mapping`
+=== `schema_evolution.processors`
 
-A mapping function from the Redpanda Connect data types to column data types in Snowflake. This mapping should result in the assignment of a string to the `root` variable with the data type for the new column in Snowflake.
+A series of processors to execute when new columns are added to the Snowflake table. You can use these processors to:
 
-The input to this mapping is an object with the value and the name of the new column. For example: `{"value": 42.3, "name":"new_data_field"}"`.
+- Run side effects when the schema evolves.
+- Enrich the message with additional information to guide the schema changes.
 
-You can also override the mapping function to customize data types.
+For example, a processor could read the schema from the schema registry that a message was produced with and use that schema to determine the data type of the new column in Snowflake.
 
-*Type*: `string`
+The input to these processors is an object with the value and name of the new column, the original message, and details of the Snowflake table the output writes to. 
 
-*Default*: 
+For example: `{"value": 42.3, "name":"new_data_field", "message": {"existing_data_field": 42, "new_data_field": "db_field_name"}, "db": MY_DATABASE", "schema": "MY_SCHEMA", "table": "MY_TABLE"}`
+
+The output from the processors must be a valid message, which contains a string that specifies the column type for the new column in Snowflake. The metadata remains the same as in the original message that triggered the schema update.
+
+*Type*: `array`
 
-```yml
-root = match this.value.type() {
-    this == "string" => "STRING"
-    this == "bytes" => "BINARY"
-    this == "number" => "DOUBLE"
-    this == "bool" => "BOOLEAN"
-    this == "timestamp" => "TIMESTAMP"
-    _ => "VARIANT"
-  }
-```
 
 === `build_options`