You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/03/06 22:16:43 UTC
[GitHub] [flink] bowenli86 opened a new pull request #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
bowenli86 opened a new pull request #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336
## What is the purpose of the change
add PostgresJDBCCatalog to connect to postgres metadata so that users can leverage it in Flink SQL and don't have to manually specify schemas.
## Brief change log
- added AbstractJDBCCatalog as common parents for all jdbc catalog impls
- added JDBCCatalog as user-facing skeleton
- added PostgresJDBCCatalog
## Verifying this change
added a few UT and IT
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (docs) docs will added later on
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395930404
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
Review comment:
added
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on issue #11336: [FLINK-16471][jdbc]
develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-601357497
@JingsongLi can you take another look?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393945465
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
Review comment:
it won't help. It's used in try-with-resource, the logic inside is complicated and won't worth extracting
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395933997
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+ } catch (SQLException e) {
+ throw new ValidationException(
+ String.format("Failed connecting to %s via JDBC.", defaultUrl), e);
+ }
+
+ LOG.info("Catalog {} established connection to {}", getName(), defaultUrl);
+ }
+
+ @Override
+ public void close() throws CatalogException {
+ LOG.info("Catalog {} closing", getName());
+ }
+
+ // ------ table factory ------
+
+ public Optional<TableFactory> getTableFactory() {
Review comment:
since it's deterministic that jdbc catalog will work with jdbc table factory, thus planner won't need to go thru service discovery. This is a short cut. When all external systems have their own catalogs, SPI will no long be necessary for table factories.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153845545",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393943598
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+@PublicEvolving
+public class JDBCCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCCatalog.class);
+
+ private final Catalog internal;
+
+ public JDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+
+ JDBCDialect dialect = JDBCDialects.get(baseUrl).get();
+
+ if (dialect instanceof JDBCDialects.PostgresDialect) {
+ internal = new PostgresCatalog(catalogName, defaultDatabase, username, pwd, baseUrl);
Review comment:
Unlike just dialect, the impl would vary a lot, e.g some db may reuse connections, some (like pg) can't. flink-jdbc is trying to give users a unified interface, thus as you said, it's internal impl and would be ok.
besides, this design is voted and approved in the FLIP
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335)
* 5bfd373640181fb6a0a8b0276ef7d03e53927f72 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395933109
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+@PublicEvolving
+public class JDBCCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCCatalog.class);
+
+ private final Catalog internal;
+
+ public JDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+
+ JDBCDialect dialect = JDBCDialects.get(baseUrl).get();
+
+ if (dialect instanceof JDBCDialects.PostgresDialect) {
+ internal = new PostgresCatalog(catalogName, defaultDatabase, username, pwd, baseUrl);
Review comment:
I see what you mean now, I'll move to a util class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152224619)
* 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393407007
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+@PublicEvolving
+public class JDBCCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCCatalog.class);
+
+ private final Catalog internal;
+
+ public JDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+
+ JDBCDialect dialect = JDBCDialects.get(baseUrl).get();
+
+ if (dialect instanceof JDBCDialects.PostgresDialect) {
+ internal = new PostgresCatalog(catalogName, defaultDatabase, username, pwd, baseUrl);
Review comment:
A internal Catalog looks weird to me...
I think this is just internal implementation.
Why not introduce something like `JDBCDialect`? PG implementation just need see what they can see.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395413626
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+@PublicEvolving
+public class JDBCCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JDBCCatalog.class);
+
+ private final Catalog internal;
+
+ public JDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+
+ JDBCDialect dialect = JDBCDialects.get(baseUrl).get();
+
+ if (dialect instanceof JDBCDialects.PostgresDialect) {
+ internal = new PostgresCatalog(catalogName, defaultDatabase, username, pwd, baseUrl);
Review comment:
I know this implementation is in FLIP.
But this if else still looks weird to me. Can we add this interface to JDBCDialect? like add method: `createCatalog(...)` in JDBCDialect. Instead need update JDBCCatalog every time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152224619)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393408226
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalogUtils.java
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Utils for {@link JDBCCatalog}.
+ */
+public class JDBCCatalogUtils {
+ public static void validateJDBCUrl(String url) {
Review comment:
This is hidden rules of `AbstractJDBCCatalog`, why not put it in.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11336: [FLINK-16471][jdbc]
develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595989134
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f (Fri Mar 06 22:18:57 UTC 2020)
**Warnings:**
* **1 pom.xml files were touched**: Check for build and licensing issues.
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393404749
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+ } catch (SQLException e) {
+ throw new ValidationException(
+ String.format("Failed connecting to %s via JDBC.", defaultUrl), e);
+ }
+
+ LOG.info("Catalog {} established connection to {}", getName(), defaultUrl);
+ }
+
+ @Override
+ public void close() throws CatalogException {
+ LOG.info("Catalog {} closing", getName());
+ }
+
+ // ------ table factory ------
+
+ public Optional<TableFactory> getTableFactory() {
Review comment:
Do we need overwrite this method, I thought this is a hacky way for hive.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153845545",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335)
* 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067)
* f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393408487
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalogUtils.java
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Utils for {@link JDBCCatalog}.
+ */
+public class JDBCCatalogUtils {
+ public static void validateJDBCUrl(String url) {
Review comment:
And I can not find any comments to explain this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11336: [FLINK-16471][jdbc]
develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152224619)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "PENDING",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393944724
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalogUtils.java
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Utils for {@link JDBCCatalog}.
+ */
+public class JDBCCatalogUtils {
+ public static void validateJDBCUrl(String url) {
Review comment:
added
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393944386
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalogUtils.java
##########
@@ -0,0 +1,34 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Utils for {@link JDBCCatalog}.
+ */
+public class JDBCCatalogUtils {
+ public static void validateJDBCUrl(String url) {
Review comment:
it's static, not associated with any object, so it's better to be in a util class
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067)
* f992a52b047e8f44480f84bfa398007e2a649bb4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393405572
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
Review comment:
I can see 5 places to new connection.
Can you extract a helper method like `run(Function<Connection, T>)`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395412120
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+ } catch (SQLException e) {
+ throw new ValidationException(
+ String.format("Failed connecting to %s via JDBC.", defaultUrl), e);
+ }
+
+ LOG.info("Catalog {} established connection to {}", getName(), defaultUrl);
+ }
+
+ @Override
+ public void close() throws CatalogException {
+ LOG.info("Catalog {} closing", getName());
+ }
+
+ // ------ table factory ------
+
+ public Optional<TableFactory> getTableFactory() {
Review comment:
Why we need this method? Just use SPI is not OK?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
}, {
"hash" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153845545",
"triggerID" : "5bfd373640181fb6a0a8b0276ef7d03e53927f72",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393408643
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/PostgresCatalog.java
##########
@@ -0,0 +1,323 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Catalog for PostgreSQL.
+ */
+@Internal
+public class PostgresCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PostgresCatalog.class);
+
+ public static final String DEFAULT_DATABASE = "postgres";
+
+ // ------ Postgres default objects that shouldn't be exposed to users ------
+
+ private static final Set<String> builtinDatabases = new HashSet<String>() {{
+ add("template0");
+ add("template1");
+ }};
+
+ private static final Set<String> builtinSchemas = new HashSet<String>() {{
+ add("pg_toast");
+ add("pg_temp_1");
+ add("pg_toast_temp_1");
+ add("pg_catalog");
+ add("information_schema");
+ }};
+
+ protected PostgresCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
+
+ // ------ databases ------
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ List<String> pgDatabases = new ArrayList<>();
+
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+
+ PreparedStatement ps = conn.prepareStatement("SELECT datname FROM pg_database;");
+
+ ResultSet rs = ps.executeQuery();
+
+ while (rs.next()) {
+ String dbName = rs.getString(1);
+ if (!builtinDatabases.contains(dbName)) {
+ pgDatabases.add(rs.getString(1));
+ }
+ }
+
+ return pgDatabases;
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed listing database in catalog %s", getName()), e);
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException, CatalogException {
+ if (listDatabases().contains(databaseName)) {
+ return new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ // ------ tables ------
+
+ @Override
+ public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+
+ // get all schemas
+ try (Connection conn = DriverManager.getConnection(baseUrl + databaseName, username, pwd)) {
+ PreparedStatement ps = conn.prepareStatement("SELECT schema_name FROM information_schema.schemata;");
+
+ ResultSet rs = ps.executeQuery();
+
+ List<String> schemas = new ArrayList<>();
+
+ while (rs.next()) {
+ String pgSchema = rs.getString(1);
+ if (!builtinSchemas.contains(pgSchema)) {
+ schemas.add(pgSchema);
+ }
+ }
+
+ List<String> tables = new ArrayList<>();
+
+ for (String schema : schemas) {
+ PreparedStatement stmt = conn.prepareStatement(
+ "SELECT * \n" +
+ "FROM information_schema.tables \n" +
+ "WHERE table_type = 'BASE TABLE' \n" +
+ " AND table_schema = ? \n" +
+ "ORDER BY table_type, table_name;");
+
+ stmt.setString(1, schema);
+
+ ResultSet rstables = stmt.executeQuery();
+
+ while (rstables.next()) {
+ // position 1 is database name, position 2 is schema name, position 3 is table name
+ tables.add(schema + "." + rstables.getString(3));
+ }
+ }
+
+ return tables;
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed listing database in catalog %s", getName()), e);
+ }
+ }
+
+ @Override
+ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ if (!tableExists(tablePath)) {
+ throw new TableNotExistException(getName(), tablePath);
+ }
+
+ PostgresTablePath pgPath = PostgresTablePath.fromFlinkTableName(tablePath.getObjectName());
+
+ try (Connection conn = DriverManager.getConnection(baseUrl + tablePath.getDatabaseName(), username, pwd)) {
+
+ PreparedStatement ps = conn.prepareStatement(
+ String.format("SELECT * FROM %s;", pgPath.getFullPath()));
+
+ ResultSetMetaData rsmd = ps.getMetaData();
+
+ String[] names = new String[rsmd.getColumnCount()];
+ DataType[] types = new DataType[rsmd.getColumnCount()];
+
+ for (int i = 1; i <= rsmd.getColumnCount(); i++) {
+ names[i - 1] = rsmd.getColumnName(i);
+ types[i - 1] = fromJDBCType(rsmd, i);
+ }
+
+ TableSchema tableSchema = new TableSchema.Builder().fields(names, types).build();
+
+ return new CatalogTableImpl(
+ tableSchema,
+ new HashMap<>(),
+ ""
+ );
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed getting table %s", tablePath.getFullName()), e);
+ }
+ }
+
+ public static final String PG_BYTEA = "bytea";
+ public static final String PG_BYTEA_ARRAY = "_bytea";
+ public static final String PG_SMALLINT = "int2";
+ public static final String PG_SMALLINT_ARRAY = "_int2";
+ public static final String PG_INTEGER = "int4";
+ public static final String PG_INTEGER_ARRAY = "_int4";
+ public static final String PG_BIGINT = "int8";
+ public static final String PG_BIGINT_ARRAY = "_int8";
+ public static final String PG_REAL = "float4";
+ public static final String PG_REAL_ARRAY = "_float4";
+ public static final String PG_DOUBLE_PRECISION = "float8";
+ public static final String PG_DOUBLE_PRECISION_ARRAY = "_float8";
+ public static final String PG_NUMERIC = "numeric";
+ public static final String PG_NUMERIC_ARRAY = "_numeric";
+ public static final String PG_BOOLEAN = "bool";
+ public static final String PG_BOOLEAN_ARRAY = "_bool";
+ public static final String PG_TIMESTAMP = "timestamp";
+ public static final String PG_TIMESTAMP_ARRAY = "_timestamp";
+ public static final String PG_TIMESTAMPTZ = "timestamptz";
+ public static final String PG_TIMESTAMPTZ_ARRAY = "_timestamptz";
+ public static final String PG_DATE = "date";
+ public static final String PG_DATE_ARRAY = "_date";
+ public static final String PG_TIME = "time";
+ public static final String PG_TIME_ARRAY = "_time";
+ public static final String PG_TEXT = "text";
+ public static final String PG_TEXT_ARRAY = "_text";
+ public static final String PG_CHAR = "bpchar";
+ public static final String PG_CHAR_ARRAY = "_bpchar";
+ public static final String PG_CHARACTER = "character";
+ public static final String PG_CHARACTER_ARRAY = "_character";
+ public static final String PG_CHARACTER_VARYING = "varchar";
+ public static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
+
+ // TODO: set precision for timestamp, timestamptz, and time
Review comment:
Why not do now?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 closed pull request #11336: [FLINK-16471][jdbc]
develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 closed pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] JingsongLi commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r395412510
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/JDBCCatalog.java
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialect;
+import org.apache.flink.api.java.io.jdbc.dialect.JDBCDialects;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
Review comment:
Comment?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f8e8521f5e342d9d8495ba7930d323d27e7ed3f Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152224619)
* 7825f1a2049defb7c15de4f2fb688799c108d01e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on issue #11336: [FLINK-16471][jdbc]
develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-601952650
> Sorry for late reply.
> Should we add SPI for catalogs?
Thanks for the review! Adding it to SPI will be in another PR.
I'll merge this PR now since all comments have been addressed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393944924
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/PostgresCatalog.java
##########
@@ -0,0 +1,323 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Catalog for PostgreSQL.
+ */
+@Internal
+public class PostgresCatalog extends AbstractJDBCCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PostgresCatalog.class);
+
+ public static final String DEFAULT_DATABASE = "postgres";
+
+ // ------ Postgres default objects that shouldn't be exposed to users ------
+
+ private static final Set<String> builtinDatabases = new HashSet<String>() {{
+ add("template0");
+ add("template1");
+ }};
+
+ private static final Set<String> builtinSchemas = new HashSet<String>() {{
+ add("pg_toast");
+ add("pg_temp_1");
+ add("pg_toast_temp_1");
+ add("pg_catalog");
+ add("information_schema");
+ }};
+
+ protected PostgresCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase, username, pwd, baseUrl);
+ }
+
+ // ------ databases ------
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ List<String> pgDatabases = new ArrayList<>();
+
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+
+ PreparedStatement ps = conn.prepareStatement("SELECT datname FROM pg_database;");
+
+ ResultSet rs = ps.executeQuery();
+
+ while (rs.next()) {
+ String dbName = rs.getString(1);
+ if (!builtinDatabases.contains(dbName)) {
+ pgDatabases.add(rs.getString(1));
+ }
+ }
+
+ return pgDatabases;
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed listing database in catalog %s", getName()), e);
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException, CatalogException {
+ if (listDatabases().contains(databaseName)) {
+ return new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ // ------ tables ------
+
+ @Override
+ public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+
+ // get all schemas
+ try (Connection conn = DriverManager.getConnection(baseUrl + databaseName, username, pwd)) {
+ PreparedStatement ps = conn.prepareStatement("SELECT schema_name FROM information_schema.schemata;");
+
+ ResultSet rs = ps.executeQuery();
+
+ List<String> schemas = new ArrayList<>();
+
+ while (rs.next()) {
+ String pgSchema = rs.getString(1);
+ if (!builtinSchemas.contains(pgSchema)) {
+ schemas.add(pgSchema);
+ }
+ }
+
+ List<String> tables = new ArrayList<>();
+
+ for (String schema : schemas) {
+ PreparedStatement stmt = conn.prepareStatement(
+ "SELECT * \n" +
+ "FROM information_schema.tables \n" +
+ "WHERE table_type = 'BASE TABLE' \n" +
+ " AND table_schema = ? \n" +
+ "ORDER BY table_type, table_name;");
+
+ stmt.setString(1, schema);
+
+ ResultSet rstables = stmt.executeQuery();
+
+ while (rstables.next()) {
+ // position 1 is database name, position 2 is schema name, position 3 is table name
+ tables.add(schema + "." + rstables.getString(3));
+ }
+ }
+
+ return tables;
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed listing database in catalog %s", getName()), e);
+ }
+ }
+
+ @Override
+ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ if (!tableExists(tablePath)) {
+ throw new TableNotExistException(getName(), tablePath);
+ }
+
+ PostgresTablePath pgPath = PostgresTablePath.fromFlinkTableName(tablePath.getObjectName());
+
+ try (Connection conn = DriverManager.getConnection(baseUrl + tablePath.getDatabaseName(), username, pwd)) {
+
+ PreparedStatement ps = conn.prepareStatement(
+ String.format("SELECT * FROM %s;", pgPath.getFullPath()));
+
+ ResultSetMetaData rsmd = ps.getMetaData();
+
+ String[] names = new String[rsmd.getColumnCount()];
+ DataType[] types = new DataType[rsmd.getColumnCount()];
+
+ for (int i = 1; i <= rsmd.getColumnCount(); i++) {
+ names[i - 1] = rsmd.getColumnName(i);
+ types[i - 1] = fromJDBCType(rsmd, i);
+ }
+
+ TableSchema tableSchema = new TableSchema.Builder().fields(names, types).build();
+
+ return new CatalogTableImpl(
+ tableSchema,
+ new HashMap<>(),
+ ""
+ );
+ } catch (Exception e) {
+ throw new CatalogException(
+ String.format("Failed getting table %s", tablePath.getFullName()), e);
+ }
+ }
+
+ public static final String PG_BYTEA = "bytea";
+ public static final String PG_BYTEA_ARRAY = "_bytea";
+ public static final String PG_SMALLINT = "int2";
+ public static final String PG_SMALLINT_ARRAY = "_int2";
+ public static final String PG_INTEGER = "int4";
+ public static final String PG_INTEGER_ARRAY = "_int4";
+ public static final String PG_BIGINT = "int8";
+ public static final String PG_BIGINT_ARRAY = "_int8";
+ public static final String PG_REAL = "float4";
+ public static final String PG_REAL_ARRAY = "_float4";
+ public static final String PG_DOUBLE_PRECISION = "float8";
+ public static final String PG_DOUBLE_PRECISION_ARRAY = "_float8";
+ public static final String PG_NUMERIC = "numeric";
+ public static final String PG_NUMERIC_ARRAY = "_numeric";
+ public static final String PG_BOOLEAN = "bool";
+ public static final String PG_BOOLEAN_ARRAY = "_bool";
+ public static final String PG_TIMESTAMP = "timestamp";
+ public static final String PG_TIMESTAMP_ARRAY = "_timestamp";
+ public static final String PG_TIMESTAMPTZ = "timestamptz";
+ public static final String PG_TIMESTAMPTZ_ARRAY = "_timestamptz";
+ public static final String PG_DATE = "date";
+ public static final String PG_DATE_ARRAY = "_date";
+ public static final String PG_TIME = "time";
+ public static final String PG_TIME_ARRAY = "_time";
+ public static final String PG_TEXT = "text";
+ public static final String PG_TEXT_ARRAY = "_text";
+ public static final String PG_CHAR = "bpchar";
+ public static final String PG_CHAR_ARRAY = "_bpchar";
+ public static final String PG_CHARACTER = "character";
+ public static final String PG_CHARACTER_ARRAY = "_character";
+ public static final String PG_CHARACTER_VARYING = "varchar";
+ public static final String PG_CHARACTER_VARYING_ARRAY = "_varchar";
+
+ // TODO: set precision for timestamp, timestamptz, and time
Review comment:
done already, removed comment
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on issue #11336: [FLINK-16471][jdbc]
develop PostgresJDBCCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on issue #11336: [FLINK-16471][jdbc] develop PostgresJDBCCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595988603
cc @KurtYoung @lirui-apache @JingsongLi @wuchong
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] bowenli86 commented on a change in pull request #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
bowenli86 commented on a change in pull request #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#discussion_r393937598
##########
File path: flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/catalog/AbstractJDBCCatalog.java
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.api.java.io.jdbc.catalog;
+
+import org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.TableFactory;
+import org.apache.flink.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/**
+ * Abstract catalog for any JDBC catalogs.
+ */
+public abstract class AbstractJDBCCatalog extends AbstractCatalog {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractJDBCCatalog.class);
+
+ protected final String username;
+ protected final String pwd;
+ protected final String baseUrl;
+ protected final String defaultUrl;
+
+ public AbstractJDBCCatalog(String catalogName, String defaultDatabase, String username, String pwd, String baseUrl) {
+ super(catalogName, defaultDatabase);
+
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(username));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(pwd));
+ checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
+
+ JDBCCatalogUtils.validateJDBCUrl(baseUrl);
+
+ this.username = username;
+ this.pwd = pwd;
+ this.baseUrl = baseUrl.endsWith("/") ? baseUrl : baseUrl + "/";
+ this.defaultUrl = baseUrl + defaultDatabase;
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ // test connection, fail early if we cannot connect to database
+ try (Connection conn = DriverManager.getConnection(defaultUrl, username, pwd)) {
+ } catch (SQLException e) {
+ throw new ValidationException(
+ String.format("Failed connecting to %s via JDBC.", defaultUrl), e);
+ }
+
+ LOG.info("Catalog {} established connection to {}", getName(), defaultUrl);
+ }
+
+ @Override
+ public void close() throws CatalogException {
+ LOG.info("Catalog {} closing", getName());
+ }
+
+ // ------ table factory ------
+
+ public Optional<TableFactory> getTableFactory() {
Review comment:
no, it's not a hack. This is the standard way to integrate table factory with catalog
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336:
[FLINK-16471][jdbc] develop PostgresCatalog
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152224619",
"triggerID" : "7f8e8521f5e342d9d8495ba7930d323d27e7ed3f",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/152384622",
"triggerID" : "7825f1a2049defb7c15de4f2fb688799c108d01e",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
}, {
"hash" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153812595",
"triggerID" : "f992a52b047e8f44480f84bfa398007e2a649bb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services