Skip to content

#13924: in Iceberg catalog, JdbcTableOperations should catch and handle PostgresSQL's exception for duplicated keys #15434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RussellSpitzer merged 1 commit into from
Feb 26, 2026
Merged

#13924: in Iceberg catalog, JdbcTableOperations should catch and handle PostgresSQL's exception for duplicated keys #15434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 2 additions & 8 deletions core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
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;
Expand Down Expand Up @@ -337,7 +336,6 @@ public List<TableIdentifier> listTables(Namespace namespace) {
JdbcUtil.namespaceToString(namespace));
}

@SuppressWarnings("checkstyle:CyclomaticComplexity")
@Override
public void renameTable(TableIdentifier from, TableIdentifier to) {
if (from.equals(to)) {
Expand All @@ -363,9 +361,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
int updatedRecords =
execute(
err -> {
// SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
if (err instanceof SQLIntegrityConstraintViolationException
|| (err.getMessage() != null && err.getMessage().contains("constraint failed"))) {
if (JdbcUtil.isConstraintViolation(err)) {
throw new AlreadyExistsException("Table already exists: %s", to);
}
},
Expand Down Expand Up @@ -713,9 +709,7 @@ public void renameView(TableIdentifier from, TableIdentifier to) {
int updatedRecords =
execute(
err -> {
// SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
if (err instanceof SQLIntegrityConstraintViolationException
|| (err.getMessage() != null && err.getMessage().contains("constraint failed"))) {
if (JdbcUtil.isConstraintViolation(err)) {
throw new AlreadyExistsException(
"Cannot rename %s to %s. View already exists", from, to);
}
Expand Down
18 changes: 6 additions & 12 deletions core/src/main/java/org/apache/iceberg/jdbc/JdbcTableOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

import java.sql.DataTruncation;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.sql.SQLNonTransientConnectionException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
Expand Down Expand Up @@ -120,14 +119,6 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
createTable(newMetadataLocation);
}

} catch (SQLIntegrityConstraintViolationException e) {

if (currentMetadataLocation() == null) {
throw new AlreadyExistsException(e, "Table already exists: %s", tableIdentifier);
} else {
throw new UncheckedSQLException(e, "Table already exists: %s", tableIdentifier);
}

} catch (SQLTimeoutException e) {
throw new UncheckedSQLException(e, "Database Connection timeout");
} catch (SQLTransientConnectionException | SQLNonTransientConnectionException e) {
Expand All @@ -137,9 +128,12 @@ public void doCommit(TableMetadata base, TableMetadata metadata) {
} catch (SQLWarning e) {
throw new UncheckedSQLException(e, "Database warning");
} catch (SQLException e) {
// SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
if (e.getMessage() != null && e.getMessage().contains("constraint failed")) {
throw new AlreadyExistsException("Table already exists: %s", tableIdentifier);
if (JdbcUtil.isConstraintViolation(e)) {
if (currentMetadataLocation() == null) {
throw new AlreadyExistsException(e, "Table already exists: %s", tableIdentifier);
} else {
throw new UncheckedSQLException(e, "Table already exists: %s", tableIdentifier);
}
}

throw new UncheckedSQLException(e, "Unknown failure");
Expand Down
8 changes: 8 additions & 0 deletions core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
Expand All @@ -45,6 +46,7 @@ final class JdbcUtil {
JdbcCatalog.PROPERTY_PREFIX + "init-catalog-tables";

static final String RETRYABLE_STATUS_CODES = "retryable_status_codes";
private static final String POSTGRES_UNIQUE_VIOLATION_SQLSTATE = "23505";

enum SchemaVersion {
V0,
Expand Down Expand Up @@ -522,6 +524,12 @@ static Properties filterAndRemovePrefix(Map<String, String> properties, String p
return result;
}

static boolean isConstraintViolation(SQLException ex) {
return ex instanceof SQLIntegrityConstraintViolationException
|| POSTGRES_UNIQUE_VIOLATION_SQLSTATE.equals(ex.getSQLState())
|| (ex.getMessage() != null && ex.getMessage().contains("constraint failed"));
}

private static int update(
boolean isTable,
SchemaVersion schemaVersion,
Expand Down
17 changes: 6 additions & 11 deletions core/src/main/java/org/apache/iceberg/jdbc/JdbcViewOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@

import java.sql.DataTruncation;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.sql.SQLNonTransientConnectionException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
Expand Down Expand Up @@ -112,13 +111,6 @@ protected void doCommit(ViewMetadata base, ViewMetadata metadata) {
createView(newMetadataLocation);
}

} catch (SQLIntegrityConstraintViolationException e) {
if (currentMetadataLocation() == null) {
throw new AlreadyExistsException(e, "View already exists: %s", viewIdentifier);
} else {
throw new UncheckedSQLException(e, "View already exists: %s", viewIdentifier);
}

} catch (SQLTimeoutException e) {
throw new UncheckedSQLException(e, "Database Connection timeout");
} catch (SQLTransientConnectionException | SQLNonTransientConnectionException e) {
Expand All @@ -128,9 +120,12 @@ protected void doCommit(ViewMetadata base, ViewMetadata metadata) {
} catch (SQLWarning e) {
throw new UncheckedSQLException(e, "Database warning");
} catch (SQLException e) {
// SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
if (e.getMessage() != null && e.getMessage().contains("constraint failed")) {
throw new AlreadyExistsException("View already exists: %s", viewIdentifier);
if (JdbcUtil.isConstraintViolation(e)) {
if (currentMetadataLocation() == null) {
throw new AlreadyExistsException(e, "View already exists: %s", viewIdentifier);
} else {
throw new UncheckedSQLException(e, "View already exists: %s", viewIdentifier);
}
}

throw new UncheckedSQLException(e, "Unknown failure");
Expand Down
67 changes: 66 additions & 1 deletion core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1083,6 +1083,7 @@ public void testCommitExceptionWithoutMessage() {
mockedStatic
.when(() -> JdbcUtil.loadTable(any(), any(), any(), any()))
.thenThrow(new SQLException());
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("Unknown failure");
Expand All @@ -1103,12 +1104,76 @@ public void testCommitExceptionWithMessage() {
mockedStatic
.when(() -> JdbcUtil.loadTable(any(), any(), any(), any()))
.thenThrow(new SQLException("constraint failed"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(AlreadyExistsException.class)
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("Table already exists: " + tableIdent);
}
}

@Test
public void testCommitExceptionWithPostgresUniqueViolation() {
TableIdentifier tableIdent = TableIdentifier.of("db", "tbl");
BaseTable table = (BaseTable) catalog.buildTable(tableIdent, SCHEMA).create();
TableOperations ops = table.operations();
TableMetadata metadataV1 = ops.current();

table.updateSchema().addColumn("n", Types.IntegerType.get()).commit();
ops.refresh();

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadTable(any(), any(), any(), any()))
.thenThrow(new SQLException("unique violation", "23505"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("Table already exists: " + tableIdent);
}
}

@Test
public void testCreateTableConstraintExceptionWithMessage() {
TableIdentifier existingIdent = TableIdentifier.of("db", "existing_tbl");
BaseTable existing = (BaseTable) catalog.buildTable(existingIdent, SCHEMA).create();
TableMetadata metadata = existing.operations().current();

TableIdentifier newIdent = TableIdentifier.of("db", "new_tbl");
TableOperations ops = catalog.newTableOps(newIdent);

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadTable(any(), any(), any(), any()))
.thenReturn(Maps.newHashMap())
.thenThrow(new SQLException("constraint failed"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(null, metadata))
.isInstanceOf(AlreadyExistsException.class)
.hasMessageStartingWith("Table already exists: " + newIdent);
}
}

@Test
public void testCreateTableConstraintExceptionWithPostgresUniqueViolation() {
TableIdentifier existingIdent = TableIdentifier.of("db", "existing_tbl2");
BaseTable existing = (BaseTable) catalog.buildTable(existingIdent, SCHEMA).create();
TableMetadata metadata = existing.operations().current();

TableIdentifier newIdent = TableIdentifier.of("db", "new_tbl2");
TableOperations ops = catalog.newTableOps(newIdent);

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadTable(any(), any(), any(), any()))
.thenReturn(Maps.newHashMap())
.thenThrow(new SQLException("unique violation", "23505"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(null, metadata))
.isInstanceOf(AlreadyExistsException.class)
.hasMessageStartingWith("Table already exists: " + newIdent);
}
}

private String createMetadataLocationViaJdbcCatalog(TableIdentifier identifier)
throws SQLException {
// temporary connection just to actually create a concrete metadata location
Expand Down
24 changes: 24 additions & 0 deletions core/src/test/java/org/apache/iceberg/jdbc/TestJdbcUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import java.nio.file.Files;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.util.Map;
import java.util.Properties;
import org.apache.iceberg.catalog.Namespace;
Expand Down Expand Up @@ -144,6 +146,28 @@ public void testV0toV1SqlStatements() throws Exception {
}
}

@Test
public void testIsConstraintViolationWithSQLIntegrityConstraintViolationException() {
assertThat(JdbcUtil.isConstraintViolation(new SQLIntegrityConstraintViolationException()))
.isTrue();
}

@Test
public void testIsConstraintViolationWithPostgresSQLState() {
assertThat(JdbcUtil.isConstraintViolation(new SQLException("unique violation", "23505")))
.isTrue();
}

@Test
public void testIsConstraintViolationWithConstraintFailedMessage() {
assertThat(JdbcUtil.isConstraintViolation(new SQLException("constraint failed"))).isTrue();
}

@Test
public void testIsConstraintViolationWithPlainSQLException() {
assertThat(JdbcUtil.isConstraintViolation(new SQLException("some other error"))).isFalse();
}

@Test
public void emptyNamespaceInIdentifier() {
assertThat(JdbcUtil.stringToTableIdentifier("", "tblName"))
Expand Down
88 changes: 87 additions & 1 deletion core/src/test/java/org/apache/iceberg/jdbc/TestJdbcViewCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ public void testCommitExceptionWithoutMessage() {
mockedStatic
.when(() -> JdbcUtil.loadView(any(), any(), any(), any()))
.thenThrow(new SQLException());
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("Unknown failure");
Expand Down Expand Up @@ -139,12 +140,97 @@ public void testCommitExceptionWithMessage() {
mockedStatic
.when(() -> JdbcUtil.loadView(any(), any(), any(), any()))
.thenThrow(new SQLException("constraint failed"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(AlreadyExistsException.class)
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("View already exists: " + identifier);
}
}

@Test
public void testCommitExceptionWithPostgresUniqueViolation() {
TableIdentifier identifier = TableIdentifier.of("namespace1", "view");
BaseView view =
(BaseView)
catalog
.buildView(identifier)
.withQuery("spark", "select * from tbl")
.withSchema(SCHEMA)
.withDefaultNamespace(Namespace.of("namespace1"))
.create();
ViewOperations ops = view.operations();
ViewMetadata metadataV1 = ops.current();

view.updateProperties().set("k1", "v1").commit();
ops.refresh();

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadView(any(), any(), any(), any()))
.thenThrow(new SQLException("unique violation", "23505"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(ops.current(), metadataV1))
.isInstanceOf(UncheckedSQLException.class)
.hasMessageStartingWith("View already exists: " + identifier);
}
}

@Test
public void testCreateViewConstraintExceptionWithMessage() {
TableIdentifier existingIdent = TableIdentifier.of("namespace1", "existing_view");
BaseView existing =
(BaseView)
catalog
.buildView(existingIdent)
.withQuery("spark", "select * from tbl")
.withSchema(SCHEMA)
.withDefaultNamespace(Namespace.of("namespace1"))
.create();
ViewMetadata metadata = existing.operations().current();

TableIdentifier newIdent = TableIdentifier.of("namespace1", "new_view");
ViewOperations ops = catalog.newViewOps(newIdent);

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadView(any(), any(), any(), any()))
.thenReturn(Maps.newHashMap())
.thenThrow(new SQLException("constraint failed"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(null, metadata))
.isInstanceOf(AlreadyExistsException.class)
.hasMessageStartingWith("View already exists: " + newIdent);
}
}

@Test
public void testCreateViewConstraintExceptionWithPostgresUniqueViolation() {
TableIdentifier existingIdent = TableIdentifier.of("namespace1", "existing_view2");
BaseView existing =
(BaseView)
catalog
.buildView(existingIdent)
.withQuery("spark", "select * from tbl")
.withSchema(SCHEMA)
.withDefaultNamespace(Namespace.of("namespace1"))
.create();
ViewMetadata metadata = existing.operations().current();

TableIdentifier newIdent = TableIdentifier.of("namespace1", "new_view2");
ViewOperations ops = catalog.newViewOps(newIdent);

try (MockedStatic<JdbcUtil> mockedStatic = Mockito.mockStatic(JdbcUtil.class)) {
mockedStatic
.when(() -> JdbcUtil.loadView(any(), any(), any(), any()))
.thenReturn(Maps.newHashMap())
.thenThrow(new SQLException("unique violation", "23505"));
mockedStatic.when(() -> JdbcUtil.isConstraintViolation(any())).thenCallRealMethod();
assertThatThrownBy(() -> ops.commit(null, metadata))
.isInstanceOf(AlreadyExistsException.class)
.hasMessageStartingWith("View already exists: " + newIdent);
}
}

@Test
public void dropViewShouldNotDropMetadataFileIfGcNotEnabled() {
TableIdentifier identifier = TableIdentifier.of("namespace1", "view");
Expand Down