diff --git a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json index 86bf1193abd9..273008f1b3ee 100644 --- a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json +++ b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "revision": 7 + "revision": 8 } diff --git a/.github/workflows/beam_PostCommit_Yaml_Xlang_Direct.yml b/.github/workflows/beam_PostCommit_Yaml_Xlang_Direct.yml index 8324c192e948..8ad2b342904b 100644 --- a/.github/workflows/beam_PostCommit_Yaml_Xlang_Direct.yml +++ b/.github/workflows/beam_PostCommit_Yaml_Xlang_Direct.yml @@ -56,12 +56,12 @@ jobs: (github.event_name == 'schedule' && github.repository == 'apache/beam') runs-on: [self-hosted, ubuntu-24.04, main] timeout-minutes: 180 - name: ${{ matrix.job_name }} (${{ matrix.job_phrase }}) + name: ${{ matrix.job_name }} (${{ matrix.job_phrase }} ${{ matrix.test_set }}) strategy: matrix: job_name: ["beam_PostCommit_Yaml_Xlang_Direct"] job_phrase: ["Run Yaml_Xlang_Direct PostCommit"] - test_set: ["data", "databases", "messaging"] + test_set: ["data", "databases", "messaging", "blueprints"] steps: - uses: actions/checkout@v7 with: diff --git a/sdks/python/apache_beam/yaml/extended_tests/blueprints/delta_lake_to_iceberg.yaml b/sdks/python/apache_beam/yaml/extended_tests/blueprints/delta_lake_to_iceberg.yaml new file mode 100644 index 000000000000..ae54d0d8d8b3 --- /dev/null +++ b/sdks/python/apache_beam/yaml/extended_tests/blueprints/delta_lake_to_iceberg.yaml @@ -0,0 +1,71 @@ +# +# 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. +# + +fixtures: + - name: DELTA_TABLE + type: "apache_beam.yaml.integration_tests.temp_delta_table" + - name: ICEBERG_FIXTURE + type: "apache_beam.yaml.integration_tests.temp_iceberg_table_with_pk" + config: + table_data: + name: "delta_lake_migration" + schema: + type: "struct" + schema-id: 0 + fields: + - { id: 1, name: "name", required: true, type: "string" } + identifier-field-ids: [1] + +pipelines: + # Pipeline 1: Read from Delta Lake and write to Iceberg + - pipeline: + type: chain + transforms: + - type: ReadFromDelta + config: + table: "{DELTA_TABLE}" + - type: WriteToIceberg + config: + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog + catalog_properties: + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" + options: + project: "apache-beam-testing" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" + + # Pipeline 2: Read from Iceberg and verify content + - pipeline: + type: chain + transforms: + - type: ReadFromIceberg + config: + table: "{ICEBERG_FIXTURE[table]}" + catalog_name: rest_catalog + catalog_properties: + type: rest + uri: "{ICEBERG_FIXTURE[api_url]}" + - type: AssertEqual + config: + elements: + - {name: "a"} + - {name: "b"} + - {name: "c"} + options: + project: "apache-beam-testing" + temp_location: "{ICEBERG_FIXTURE[temp_dir]}" diff --git a/sdks/python/build.gradle b/sdks/python/build.gradle index b50660978b2a..6549aa394823 100644 --- a/sdks/python/build.gradle +++ b/sdks/python/build.gradle @@ -174,7 +174,7 @@ tasks.register("postCommitYamlIntegrationTests") { dependsOn ":sdks:java:io:debezium:expansion-service:shadowJar" doLast { - def testSetInput = project.findProperty('yamlTestSet') ?: 'data,databases,messaging' + def testSetInput = project.findProperty('yamlTestSet') ?: 'data,databases,messaging,blueprints' def testSetsToRun = testSetInput.tokenize(',').collect { it.trim() }.findAll { !it.isEmpty() } testSetsToRun.each { currentTestSet -> def test_files_dir @@ -189,8 +189,11 @@ tasks.register("postCommitYamlIntegrationTests") { case 'messaging': test_files_dir = 'extended_tests/messaging' break + case 'blueprints': + test_files_dir = 'extended_tests/blueprints' + break default: - throw StopExecutionException("Unknown yamlTestSet: ${currentTestSet}. Must be one of 'data', 'databases', or 'messaging'.") + throw StopExecutionException("Unknown yamlTestSet: ${currentTestSet}. Must be one of 'data', 'databases', 'messaging', or 'blueprints'.") } exec { executable 'sh'