flink-connector-jdbc-db2/src/main/java/org/apache/flink/connector/jdbc/db2/database/lineage/DB2LocationExtractor.java

@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.db2.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocation;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.OverrideJdbcLocationExtractor;
+
+import java.net.URISyntaxException;
+import java.util.Properties;
+
+/**
+ * Implementation of {@link JdbcLocationExtractor} for DB2.
+ *
+ * @see <a href="https://www.ibm.com/docs/en/db2woc?topic=programmatically-jdbc">DB2 URL Format</a>
+ */
+@Internal
+public class DB2LocationExtractor implements JdbcLocationExtractor {
+
+    private JdbcLocationExtractor delegate() {
+        return new OverrideJdbcLocationExtractor("db2");
+    }
+
+    @Override
+    public boolean isDefinedAt(String jdbcUri) {
+        return delegate().isDefinedAt(jdbcUri);
+    }
+
+    @Override
+    public JdbcLocation extract(String rawUri, Properties properties) throws URISyntaxException {
+        return delegate().extract(rawUri, properties);
+    }
+}

flink-connector-jdbc-db2/src/main/java/org/apache/flink/connector/jdbc/db2/database/lineage/DB2LocationExtractorFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.db2.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory;
+
+/** Implementation of {@link JdbcLocationExtractorFactory} for DB2. */
+@Internal
+public class DB2LocationExtractorFactory implements JdbcLocationExtractorFactory {
+
+    @Override
+    public JdbcLocationExtractor createExtractor() {
+        return new DB2LocationExtractor();
+    }
+}

flink-connector-jdbc-db2/src/main/resources/META-INF/services/org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.connector.jdbc.db2.database.lineage.DB2LocationExtractorFactory

flink-connector-jdbc-oceanbase/src/main/java/org/apache/flink/connector/jdbc/oceanbase/database/lineage/OceanBaseLocationExtractor.java

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.oceanbase.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocation;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.OverrideJdbcLocationExtractor;
+
+import java.net.URISyntaxException;
+import java.util.Properties;
+
+/**
+ * Implementation of {@link JdbcLocationExtractor} for OceanBase.
+ *
+ * @see <a
+ *     href="https://en.oceanbase.com/docs/common-oceanbase-connector-j-en-10000000000381036">OceanBase
+ *     URL Format</a>
+ */
+@Internal
+public class OceanBaseLocationExtractor implements JdbcLocationExtractor {
+
+    private static final String PROTOCOL_PART = "^[\\w+:]+://";
+
+    private JdbcLocationExtractor delegate() {
+        return new OverrideJdbcLocationExtractor("oceanbase");
+    }
+
+    @Override
+    public boolean isDefinedAt(String jdbcUri) {
+        return delegate().isDefinedAt(jdbcUri);
+    }
+
+    @Override
+    public JdbcLocation extract(String rawUri, Properties properties) throws URISyntaxException {
+        // Schema could be 'oceanbase', 'oceanbase:hamode'. Convert it to 'mysql'
+        String normalizedUri = rawUri.replaceFirst(PROTOCOL_PART, "oceanbase://");
+        return delegate().extract(normalizedUri, properties);
+    }
+}

flink-connector-jdbc-oceanbase/src/main/java/org/apache/flink/connector/jdbc/oceanbase/database/lineage/OceanBaseLocationExtractorFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.oceanbase.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory;
+
+/** Implementation of {@link JdbcLocationExtractorFactory} for Mysql. */
+@Internal
+public class OceanBaseLocationExtractorFactory implements JdbcLocationExtractorFactory {
+
+    @Override
+    public JdbcLocationExtractor createExtractor() {
+        return new OceanBaseLocationExtractor();
+    }
+}

flink-connector-jdbc-oceanbase/src/main/resources/META-INF/services/org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.connector.jdbc.oceanbase.database.lineage.OceanBaseLocationExtractorFactory

flink-connector-jdbc-trino/src/main/java/org/apache/flink/connector/jdbc/trino/database/lineage/TrinoLocationExtractor.java

@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.trino.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocation;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.OverrideJdbcLocationExtractor;
+
+import java.net.URISyntaxException;
+import java.util.Properties;
+
+/**
+ * Implementation of {@link JdbcLocationExtractor} for Trino.
+ *
+ * @see <a href="https://trino.io/docs/current/client/jdbc.html">Trino URL Format</a>
+ */
+@Internal
+public class TrinoLocationExtractor implements JdbcLocationExtractor {
+
+    private JdbcLocationExtractor delegate() {
+        return new OverrideJdbcLocationExtractor("postgres");
+    }
+
+    @Override
+    public boolean isDefinedAt(String jdbcUri) {
+        return delegate().isDefinedAt(jdbcUri);
+    }
+
+    @Override
+    public JdbcLocation extract(String rawUri, Properties properties) throws URISyntaxException {
+        return delegate().extract(rawUri, properties);
+    }
+}

flink-connector-jdbc-trino/src/main/java/org/apache/flink/connector/jdbc/trino/database/lineage/TrinoLocationExtractorFactory.java

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.connector.jdbc.trino.database.lineage;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractor;
+import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory;
+
+/** Implementation of {@link JdbcLocationExtractorFactory} for Postgres. */
+@Internal
+public class TrinoLocationExtractorFactory implements JdbcLocationExtractorFactory {
+
+    @Override
+    public JdbcLocationExtractor createExtractor() {
+        return new TrinoLocationExtractor();
+    }
+}

flink-connector-jdbc-trino/src/main/resources/META-INF/services/org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory

@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.connector.jdbc.trino.database.lineage.TrinoLocationExtractorFactory