Merge remote-tracking branch 'mygit/feature/postgresql-catalog-uuid-type-support' into feature/postgresql-catalog-uuid-type-support

# Conflicts:
#	flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresCatalogTest.java
#	flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/catalog/PostgresCatalogTestBase.java

Author: och5351

.github/workflows/push_pr.yml

@@ -28,7 +28,7 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [2.0.0]
+        flink: [2.0.1, 2.1.1]
         jdk: [ '17' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:

.github/workflows/stale.yml

@@ -0,0 +1,61 @@
+# 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 workflow labels and then closes stale PRs that haven't seen attention
+# for several months.
+
+name: Stale PRs
+on:
+  schedule:
+    - cron: '15 6 * * *' # Run once a day at 6:15 UTC
+  workflow_dispatch:
+    inputs:
+      operationsPerRun:
+        description: 'Max GitHub API operations'
+        required: true
+        default: 20
+        type: number
+
+permissions:
+  issues: write
+  pull-requests: write
+  actions: write
+
+jobs:
+  stale:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/stale@v9
+        with:
+          operations-per-run: ${{ inputs.operationsPerRun || 500 }}
+          ascending: true
+          days-before-stale: 90
+          days-before-close: 30
+          stale-pr-label: 'stale'
+          stale-pr-message: |
+            This PR is being marked as stale since it has not had any activity in the last 90 days. 
+            If you would like to keep this PR alive, please leave a comment asking for a review. 
+            If the PR has merge conflicts, update it with the latest from the base branch.
+
+            If you are having difficulty finding a reviewer, please reach out to the 
+            community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
+
+            If this PR is no longer valid or desired, please feel free to close it. 
+            If no activity occurs in the next 30 days, it will be automatically closed.
+          close-pr-label: 'closed-stale'
+          close-pr-message: |
+            This PR has been closed since it has not had any activity in 120 days. 
+            If you feel like this was a mistake, or you would like to continue working on it, 
+            please feel free to re-open the PR and ask for a review.

.github/workflows/weekly.yml

@@ -34,21 +34,17 @@ jobs:
           jdk: '17',
           branch: main
         }, {
-          flink: 2.0-SNAPSHOT,
+          flink: 2.0.1,
           jdk: '17',
-          branch: main
+          branch: v4.0
         }, {
-          flink: 2.0.0,
-          jdk: '17',
-          branch: main
-        }, {
-          flink: 1.19.1,
+          flink: 1.20.3,
           jdk: '8, 11, 17, 21',
-          branch: v3.2
+          branch: v3.3
         }, {
-          flink: 1.18.0,
-          jdk: '8, 11, 17',
-          branch: v3.1
+          flink: 1.19.3,
+          jdk: '8, 11, 17, 21',
+          branch: v3.3
         }]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:

.gitignore

@@ -7,6 +7,7 @@ scalastyle-output.xml
 .metadata
 .settings
 .project
+.java-version
 .version.properties
 filter.properties
 logs.zip
@@ -35,4 +36,5 @@ out/
 tools/flink
 tools/flink-*
 tools/releasing/release
-tools/japicmp-output
+tools/japicmp-output
+.vscode/

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "tools/releasing/shared"]
 	path = tools/releasing/shared
-	url = https://github.com/apache/flink-connector-shared-utils
-    branch = release_utils
+	url = git@github.com:apache/flink-connector-shared-utils.git
+        branch = release_utils

flink-connector-jdbc-core/archunit-violations/4bca2274-65a9-4a61-81ef-767d58233ea0

@@ -0,0 +1 @@
+Method <org.apache.flink.connector.jdbc.core.datastream.source.JdbcSource.getLineageVertex()> calls method <org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator.getSqlTemplate()> in (JdbcSource.java:215)

flink-connector-jdbc-core/archunit-violations/d45c3af5-52c6-45fd-9926-75e75e77473a

@@ -0,0 +1,4 @@
+org.apache.flink.connector.jdbc.lineage.DefaultJdbcExtractor.extract(java.lang.String, java.util.Properties): Returned leaf type org.apache.flink.connector.jdbc.lineage.JdbcLocation does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
+org.apache.flink.connector.jdbc.lineage.JdbcLocation$Builder.build(): Returned leaf type org.apache.flink.connector.jdbc.lineage.JdbcLocation does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
+org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor.extract(java.lang.String, java.util.Properties): Returned leaf type org.apache.flink.connector.jdbc.lineage.JdbcLocation does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
+org.apache.flink.connector.jdbc.lineage.OverrideJdbcLocationExtractor.extract(java.lang.String, java.util.Properties): Returned leaf type org.apache.flink.connector.jdbc.lineage.JdbcLocation does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated

flink-connector-jdbc-core/pom.xml

@@ -56,6 +56,16 @@ under the License.
             <optional>true</optional>
         </dependency>
 
+        <dependency>
+            <groupId>io.openlineage</groupId>
+            <artifactId>openlineage-sql-java</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.openlineage</groupId>
+            <artifactId>openlineage-java</artifactId>
+        </dependency>
+
         <!-- Tests -->
 
         <dependency>

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/JdbcInputFormat.java

@@ -24,17 +24,23 @@
 import org.apache.flink.api.common.io.InputFormat;
 import org.apache.flink.api.common.io.RichInputFormat;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
+import org.apache.flink.api.connector.source.Boundedness;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.connector.jdbc.core.datastream.source.JdbcSource;
 import org.apache.flink.connector.jdbc.core.datastream.source.JdbcSourceBuilder;
 import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
 import org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
+import org.apache.flink.connector.jdbc.lineage.DefaultTypeDatasetFacet;
+import org.apache.flink.connector.jdbc.lineage.LineageUtils;
 import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
 import org.apache.flink.core.io.GenericInputSplit;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.core.io.InputSplitAssigner;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 
@@ -53,6 +59,8 @@
 import java.sql.Time;
 import java.sql.Timestamp;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
 
 /**
  * InputFormat to read data from a database and generate Rows. The InputFormat has to be configured
@@ -107,7 +115,7 @@
 @Deprecated
 @Experimental
 public class JdbcInputFormat extends RichInputFormat<Row, InputSplit>
-        implements ResultTypeQueryable<Row> {
+        implements LineageVertexProvider, ResultTypeQueryable<Row> {
 
     protected static final long serialVersionUID = 2L;
     protected static final Logger LOG = LoggerFactory.getLogger(JdbcInputFormat.class);
@@ -344,6 +352,19 @@ public static JdbcInputFormatBuilder buildJdbcInputFormat() {
         return new JdbcInputFormatBuilder();
     }
 
+    @Override
+    public LineageVertex getLineageVertex() {
+        DefaultTypeDatasetFacet defaultTypeDatasetFacet =
+                new DefaultTypeDatasetFacet(getProducedType());
+        Optional<String> nameOpt = LineageUtils.tableNameOf(queryTemplate, true);
+        String namespace = LineageUtils.namespaceOf(connectionProvider);
+        LineageDataset dataset =
+                LineageUtils.datasetOf(
+                        nameOpt.orElse(""), namespace, Arrays.asList(defaultTypeDatasetFacet));
+        return LineageUtils.sourceLineageVertexOf(
+                Boundedness.BOUNDED, Collections.singleton(dataset));
+    }
+
     /** Builder for {@link JdbcInputFormat}. */
     public static class JdbcInputFormatBuilder {
         private final JdbcConnectionOptions.JdbcConnectionOptionsBuilder connOptionsBuilder;

flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/JdbcSink.java

@@ -37,11 +37,16 @@
 import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
 import org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement;
 import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer;
+import org.apache.flink.connector.jdbc.lineage.LineageUtils;
 import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.lineage.LineageDataset;
+import org.apache.flink.streaming.api.lineage.LineageVertex;
+import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
 
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Optional;
 
 /**
  * Flink Sink to produce data into a jdbc database.
@@ -50,7 +55,8 @@
  */
 @PublicEvolving
 public class JdbcSink<IN>
-        implements Sink<IN>,
+        implements LineageVertexProvider,
+                Sink<IN>,
                 SupportsWriterState<IN, JdbcWriterState>,
                 SupportsCommitter<JdbcCommitable> {
 
@@ -120,4 +126,13 @@ public JdbcWriter<IN> restoreWriter(
     public SimpleVersionedSerializer<JdbcWriterState> getWriterStateSerializer() {
         return new JdbcWriterStateSerializer();
     }
+
+    @Override
+    public LineageVertex getLineageVertex() {
+        Optional<String> nameOpt = LineageUtils.tableNameOf(queryStatement.query(), false);
+        String namespace = LineageUtils.namespaceOf(connectionProvider);
+        LineageDataset dataset =
+                LineageUtils.datasetOf(nameOpt.orElse(""), namespace, Collections.emptyList());
+        return LineageUtils.lineageVertexOf(Collections.singleton(dataset));
+    }
 }