Skip to content

Commit

Permalink
[Feature][transform-v2] jsonpath support map array type
Browse files Browse the repository at this point in the history
  • Loading branch information
nijiahui committed Jan 23, 2025
1 parent 2bfb97e commit e9c6285
Show file tree
Hide file tree
Showing 3 changed files with 95 additions and 0 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,9 @@ private static SeaTunnelDataType<?> parseArrayType(String field, String columnSt
return ArrayType.FLOAT_ARRAY_TYPE;
case DOUBLE:
return ArrayType.DOUBLE_ARRAY_TYPE;
case MAP:
return new ArrayType<>(
MapType.class, new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE));
default:
throw CommonError.unsupportedDataType("SeaTunnel", genericType, field);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,4 +59,11 @@ public void testErrorHandleWay(TestContainer container) throws Exception {
container.executeJob("/json_path_transform/json_path_with_error_handle_way.conf");
Assertions.assertEquals(0, execResult.getExitCode());
}

@TestTemplate
public void testArrayType(TestContainer container) throws Exception {
Container.ExecResult execResult =
container.executeJob("/json_path_transform/json_path_array_map.conf");
Assertions.assertEquals(0, execResult.getExitCode());
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
#
# 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.
#

######
###### This config file is a demonstration of streaming processing in seatunnel config
######

env {
job.mode = "BATCH"
}

source {
FakeSource {
plugin_output = "fake"
row.num = 100
string.fake.mode = "template"
string.template=["{"data":{"c_map_array":[{"c_string":"this is a string","c_boolean": "true","c_integer": "42"},{"c_string":"this is a string","c_boolean": "true","c_integer": "42"}]}}"]
schema = {
fields {
data = "string"
}
}
}
}

transform {
JsonPath {
plugin_input = "fake"
plugin_output = "fake1"
columns = [
{
"src_field" = "data"
"path" = "$.data.c_map_array"
"dest_field" = "c_map_array_1"
"dest_type" = "array<map<string, string>>"
}
]
}
Sql {
plugin_input = "fake1"
plugin_output = "fake2"
query = "select c_map_array_1 from dual"
}
}

sink {
Assert {
plugin_input = "fake2"
rules =
{
row_rules = [
{
rule_type = MIN_ROW
rule_value = 100
}
],
field_rules = [
{
field_name = c_map_array_1
field_type = "array<map<string, string>>"
field_value = [
{
rule_type = NOT_NULL
equals_to = [{c_string=this is a string, c_boolean=true, c_integer=42}, {c_string=this is a string, c_boolean=true, c_integer=42}]
}
]
}
]
}
}
}

0 comments on commit e9c6285

Please sign in to comment.