jackye1995 commented on a change in pull request #1870:
URL: https://github.com/apache/iceberg/pull/1870#discussion_r556306959
##########
File path: core/src/main/java/org/apache/iceberg/ClientPool.java
##########
@@ -36,7 +36,7 @@
private volatile int currentSize;
private boolean closed;
- ClientPool(int poolSize, Class<? extends E> reconnectExc) {
+ protected ClientPool(int poolSize, Class<? extends E> reconnectExc) {
Review comment:
nit: unnecessary keyword `protected`
##########
File path:
api/src/main/java/org/apache/iceberg/exceptions/UncheckedSQLException.java
##########
@@ -0,0 +1,29 @@
+/*
+ * 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.iceberg.exceptions;
+
+import java.sql.SQLException;
+
+public class UncheckedSQLException extends RuntimeException {
Review comment:
I think this exception is specific to JDBC catalog and should be in the
core jdbc package path.
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java
##########
@@ -0,0 +1,404 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.io.Closeable;
+import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.sql.SQLNonTransientConnectionException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseMetastoreCatalog;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JdbcCatalog extends BaseMetastoreCatalog implements Configurable,
SupportsNamespaces, Closeable {
+
+ protected static final String JDBC_PARAM_PREFIX = "connection.parameter.";
Review comment:
1. This needs to be public, so that users can import this variable to
set their custom properties.
2. to be consistent with the name `CatalogProperties`, this should be named
`PROPERTY_PREFIX`, so that users can use `JdbcCatalog.PROPERTY_PREFIX`.
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.DataTruncation;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.sql.SQLNonTransientConnectionException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.sql.SQLWarning;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class JdbcTableOperations extends BaseMetastoreTableOperations {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcTableOperations.class);
+ private final String catalogName;
+ private final TableIdentifier tableIdentifier;
+ private final FileIO fileIO;
+ private final JdbcClientPool connections;
+
+ protected JdbcTableOperations(JdbcClientPool dbConnPool, FileIO fileIO,
String catalogName,
+ TableIdentifier tableIdentifier) {
+ this.catalogName = catalogName;
+ this.tableIdentifier = tableIdentifier;
+ this.fileIO = fileIO;
+ this.connections = dbConnPool;
+ }
+
+ @Override
+ public void doRefresh() {
+ Map<String, String> table;
+
+ try {
+ table = getTable();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted during refresh", e);
+ } catch (SQLException e) {
+ // unknown exception happened when getting table from catalog
+ throw new UncheckedSQLException(String.format("Failed to get table from
catalog %s.%s", catalogName,
+ tableIdentifier), e);
+ }
+
+ // Table not exists AND currentMetadataLocation is not NULL!
+ if (table.isEmpty() && currentMetadataLocation() != null) {
+ throw new NoSuchTableException("Failed to get table from catalog %s.%s!"
+
+ " maybe another process deleted it!", catalogName, tableIdentifier);
+ }
+
+ // Table not exists in the catalog! metadataLocation is null here!
+ if (table.isEmpty()) {
+ refreshFromMetadataLocation(null);
+ return;
+ }
+
+ // Table exists but metadataLocation is null
+ if (table.getOrDefault("metadata_location", null) == null) {
Review comment:
"metadata_location" should be a static variable
##########
File path: build.gradle
##########
@@ -231,6 +231,7 @@ project(':iceberg-core') {
exclude group: 'org.slf4j', module: 'slf4j-log4j12'
}
+ testCompile "com.h2database:h2"
Review comment:
The MPL license is a copyleft license, which means that in principle
people are not allowed to distribute code that is under the MPL-2.0 license
under different terms. The GPL licenses (including LGPL and AGPL) require that
the entire application is distributed under the terms of the GPL license.
I think we cannot use it. Could you switch to use
https://mvnrepository.com/artifact/org.xerial/sqlite-jdbc?
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.DataTruncation;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.sql.SQLNonTransientConnectionException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.sql.SQLWarning;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class JdbcTableOperations extends BaseMetastoreTableOperations {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcTableOperations.class);
+ private final String catalogName;
+ private final TableIdentifier tableIdentifier;
+ private final FileIO fileIO;
+ private final JdbcClientPool connections;
+
+ protected JdbcTableOperations(JdbcClientPool dbConnPool, FileIO fileIO,
String catalogName,
+ TableIdentifier tableIdentifier) {
+ this.catalogName = catalogName;
+ this.tableIdentifier = tableIdentifier;
+ this.fileIO = fileIO;
+ this.connections = dbConnPool;
+ }
+
+ @Override
+ public void doRefresh() {
+ Map<String, String> table;
+
+ try {
+ table = getTable();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted during refresh", e);
+ } catch (SQLException e) {
+ // unknown exception happened when getting table from catalog
+ throw new UncheckedSQLException(String.format("Failed to get table from
catalog %s.%s", catalogName,
+ tableIdentifier), e);
+ }
+
+ // Table not exists AND currentMetadataLocation is not NULL!
+ if (table.isEmpty() && currentMetadataLocation() != null) {
+ throw new NoSuchTableException("Failed to get table from catalog %s.%s!"
+
+ " maybe another process deleted it!", catalogName, tableIdentifier);
+ }
+
+ // Table not exists in the catalog! metadataLocation is null here!
+ if (table.isEmpty()) {
+ refreshFromMetadataLocation(null);
+ return;
+ }
+
+ // Table exists but metadataLocation is null
+ if (table.getOrDefault("metadata_location", null) == null) {
+ throw new RuntimeException(String.format("Failed to get metadata
location if the table %s.%s", catalogName,
+ tableIdentifier));
+ }
+
+ refreshFromMetadataLocation(table.get("metadata_location"));
+ }
+
+ @Override
+ public void doCommit(TableMetadata base, TableMetadata metadata) {
Review comment:
method too long, prefer to separate to smaller methods like
`updateTable` and `createTable` based on exist condition.
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcClientPool.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLNonTransientConnectionException;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+
+public class JdbcClientPool extends ClientPool<Connection, SQLException> {
Review comment:
This do not need to be public
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.DataTruncation;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.sql.SQLNonTransientConnectionException;
+import java.sql.SQLTimeoutException;
+import java.sql.SQLTransientConnectionException;
+import java.sql.SQLWarning;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class JdbcTableOperations extends BaseMetastoreTableOperations {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcTableOperations.class);
+ private final String catalogName;
+ private final TableIdentifier tableIdentifier;
+ private final FileIO fileIO;
+ private final JdbcClientPool connections;
+
+ protected JdbcTableOperations(JdbcClientPool dbConnPool, FileIO fileIO,
String catalogName,
+ TableIdentifier tableIdentifier) {
+ this.catalogName = catalogName;
+ this.tableIdentifier = tableIdentifier;
+ this.fileIO = fileIO;
+ this.connections = dbConnPool;
+ }
+
+ @Override
+ public void doRefresh() {
+ Map<String, String> table;
+
+ try {
+ table = getTable();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted during refresh", e);
+ } catch (SQLException e) {
+ // unknown exception happened when getting table from catalog
+ throw new UncheckedSQLException(String.format("Failed to get table from
catalog %s.%s", catalogName,
Review comment:
nit: `Failed to get table %s from catalog %s, tableIdentifier,
catalogName` sounds better, also for other similar error messages below.
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcClientPool.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLNonTransientConnectionException;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+
+public class JdbcClientPool extends ClientPool<Connection, SQLException> {
+
+ private final String dbUrl;
+ private final Map<String, String> properties;
+
+ JdbcClientPool(String dbUrl, Map<String, String> props) {
+
this(Integer.parseInt(props.getOrDefault(CatalogProperties.HIVE_CLIENT_POOL_SIZE,
+ String.valueOf(CatalogProperties.HIVE_CLIENT_POOL_SIZE_DEFAULT))),
dbUrl, props);
+ }
+
+ public JdbcClientPool(int poolSize, String dbUrl, Map<String, String> props)
{
+ super(poolSize, SQLNonTransientConnectionException.class);
+ properties = props;
+ this.dbUrl = dbUrl;
+ }
+
+ @Override
+ protected Connection newClient() {
+ try {
+ Properties dbProps = new Properties();
+ properties.forEach((key, value) ->
dbProps.put(key.replace(JdbcCatalog.JDBC_PARAM_PREFIX, ""), value));
+ return DriverManager.getConnection(dbUrl, dbProps);
+ } catch (SQLException e) {
+ throw new UncheckedSQLException("Failed to connect: " + dbUrl, e);
+ }
+ }
+
+ @Override
+ protected Connection reconnect(Connection client) {
+ close(client);
+ return newClient();
+ }
+
+ @Override
+ protected void close(Connection client) {
+ try {
+ client.close();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to connect to database!", e);
Review comment:
why not use your `UncheckedSQLException`?
##########
File path: core/src/main/java/org/apache/iceberg/jdbc/JdbcClientPool.java
##########
@@ -0,0 +1,73 @@
+/*
+ * 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.iceberg.jdbc;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLNonTransientConnectionException;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.exceptions.UncheckedSQLException;
+
+public class JdbcClientPool extends ClientPool<Connection, SQLException> {
+
+ private final String dbUrl;
+ private final Map<String, String> properties;
+
+ JdbcClientPool(String dbUrl, Map<String, String> props) {
+
this(Integer.parseInt(props.getOrDefault(CatalogProperties.HIVE_CLIENT_POOL_SIZE,
+ String.valueOf(CatalogProperties.HIVE_CLIENT_POOL_SIZE_DEFAULT))),
dbUrl, props);
+ }
+
+ public JdbcClientPool(int poolSize, String dbUrl, Map<String, String> props)
{
Review comment:
This do not need to be public
----------------------------------------------------------------
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:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
