#
# 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 {
  execution.parallelism = 1
  job.mode = "BATCH"
  checkpoint.interval = 10000
}

source {
  FakeSource {
    result_table_name = "fake"
    schema = {
      fields {
        id = "int"
        name = "string"
        c1 = "string"
        c2 = "string"
        c3 = "string"
        c4 = "string"
        c5 = "int"
        c6 = "string"
        c7 = "string"
        c8 = "string"
        c9 = "string"
        c10 = "string"
        c11 = "timestamp"
      }
    }
    rows = [
      {fields = [1, "Joy Ding", "A", "b", "&", "&^^$wef9", 98, "0037", "7", "*Joy_Ding@s.com*", "Joy_WWWDing@s.com", "2020-10-01", "2022-12-12T23:34:45"], kind = INSERT}
    ]
  }
}

transform {
  sql {
    source_table_name = "fake"
    result_table_name = "fake1"
    query = "select ascii(c1) as c1_1, ascii(c2) as c2_1, bit_length(c4) as c4_1, length(c4) as c4_2, octet_length(c4) as c4_3, char(c5) as c5_1, concat(c1,id,'!') as c1_2, hextoraw(c6) as c6_1, rawtohex(c7) as c7_1, insert(name,2,2,'**') as name1, lower(name) as name2, upper(name) as name3, left(name, 3) as name4, right(name, 4) as name5, lpad(name, 10, '*') as name6, rpad(name, 10, '*') as name7, ltrim(c8, '*') as c8_1, rtrim(c8, '*') as c8_2, trim(c8, '*') as c8_3, regexp_replace(c9, 'w+', 'W', 'i') as c9_1, regexp_like(name, '[A-Z ]*', 'i') as name8, regexp_substr(c10, '\\d{4}') as c10_1, regexp_substr(c10, '(\\d{4})-(\\d{2})-(\\d{2})', 1, 1, null, 2) as c10_2, repeat(name||' ',3) as name9, replace(name,' ','_') as name10, soundex(name) as name11, name || space(3) as name12, substring(name, 1, 3) as name13, to_char(id) as id1, to_char(c11,'yyyy-MM-dd') as c11_1, translate(name, 'ing', 'ING') as name14, des_decrypt('1234567890', des_encrypt('1234567890', name)) as name15 from fake"
  }
}

sink {
  Console {
    source_table_name = "fake1"
  }
  Assert {
    source_table_name = "fake1"
    rules = {
      field_rules = [
        {
          field_name = "c1_1"
          field_type = "int"
          field_value = [
            {equals_to = 65}
          ]
        },
        {
          field_name = "c2_1"
          field_type = "int"
          field_value = [
            {equals_to = 98}
          ]
        },
        {
          field_name = "c4_1"
          field_type = "long"
          field_value = [
            {equals_to = 64}
          ]
        },
        {
          field_name = "c4_2"
          field_type = "long"
          field_value = [
            {equals_to = 8}
          ]
        },
        {
          field_name = "c4_3"
          field_type = "long"
          field_value = [
            {equals_to = 8}
          ]
        },
        {
          field_name = "c5_1"
          field_type = "string"
          field_value = [
            {equals_to = "b"}
          ]
        },
        {
          field_name = "c1_2"
          field_type = "string"
          field_value = [
            {equals_to = "A1!"}
          ]
        },
        {
          field_name = "c6_1"
          field_type = "string"
          field_value = [
            {equals_to = "7"}
          ]
        },
        {
          field_name = "c7_1"
          field_type = "string"
          field_value = [
            {equals_to = "0037"}
          ]
        },
        {
          field_name = "name1"
          field_type = "string"
          field_value = [
            {equals_to = "J** Ding"}
          ]
        },
        {
          field_name = "name2"
          field_type = "string"
          field_value = [
            {equals_to = "joy ding"}
          ]
        },
        {
          field_name = "name3"
          field_type = "string"
          field_value = [
            {equals_to = "JOY DING"}
          ]
        },
        {
          field_name = "name4"
          field_type = "string"
          field_value = [
            {equals_to = "Joy"}
          ]
        },
        {
          field_name = "name5"
          field_type = "string"
          field_value = [
            {equals_to = "Ding"}
          ]
        },
        {
          field_name = "name6"
          field_type = "string"
          field_value = [
            {equals_to = "**Joy Ding"}
          ]
        },
        {
          field_name = "name7"
          field_type = "string"
          field_value = [
            {equals_to = "Joy Ding**"}
          ]
        },
        {
          field_name = "c8_1"
          field_type = "string"
          field_value = [
            {equals_to = "Joy_Ding@s.com*"}
          ]
        },
        {
          field_name = "c8_2"
          field_type = "string"
          field_value = [
            {equals_to = "*Joy_Ding@s.com"}
          ]
        },
        {
          field_name = "c8_3"
          field_type = "string"
          field_value = [
            {equals_to = "Joy_Ding@s.com"}
          ]
        },
        {
          field_name = "c9_1"
          field_type = "string"
          field_value = [
            {equals_to = "Joy_WDing@s.com"}
          ]
        },
        {
          field_name = "name8"
          field_type = "boolean"
          field_value = [
            {equals_to = true}
          ]
        },
        {
          field_name = "c10_1"
          field_type = "string"
          field_value = [
            {equals_to = "2020"}
          ]
        },
        {
          field_name = "c10_2"
          field_type = "string"
          field_value = [
            {equals_to = "10"}
          ]
        },
        {
          field_name = "name9"
          field_type = "string"
          field_value = [
            {equals_to = "Joy Ding Joy Ding Joy Ding "}
          ]
        },
        {
          field_name = "name10"
          field_type = "string"
          field_value = [
            {equals_to = "Joy_Ding"}
          ]
        },
        {
          field_name = "name11"
          field_type = "string"
          field_value = [
            {equals_to = "J352"}
          ]
        },
        {
          field_name = "name12"
          field_type = "string"
          field_value = [
            {equals_to = "Joy Ding   "}
          ]
        },
        {
          field_name = "name13"
          field_type = "string"
          field_value = [
            {equals_to = "Joy"}
          ]
        },
        {
          field_name = "id1"
          field_type = "string"
          field_value = [
            {equals_to = "1"}
          ]
        },
        {
          field_name = "c11_1"
          field_type = "string"
          field_value = [
            {equals_to = "2022-12-12"}
          ]
        },
        {
          field_name = "name14"
          field_type = "string"
          field_value = [
            {equals_to = "Joy DING"}
          ]
        },
        {
          field_name = "name15"
          field_type = "string"
          field_value = [
            {equals_to = "Joy Ding"}
          ]
        }
      ]
    }
  }
}