Fault Tolerant Execution for PostgreSQL and MySQL connectors #14445

mwd410 · 2022-10-03T20:12:41Z

Description

This PR primarily introduces fault tolerant writes for Postgres and MySQL. To do this, the write process is changed thusly:

Add a ConstructorPageSinkId to trino-spi. The implementation of this is PageSinkId.
1. The JdbcPageSink takes a ConstructorPageSinkId to use as a long identifier. We chose long instead of a string for the fast joins.
2. The identifier (see fromTaskId in PageSinkId), is calculated thusly:
  1. The 4 bytes from the StageId's id
  2. The bottom 3 bytes of the partitionId - this affords us ~16 million partitions
  3. The bottom byte of the attemptId - this affords us 256 attempts
When creating the temporary staging table the JdbcPageSinks write to, we subsequently add a trino_page_sink_id column to that temporary table.
The PageSink will include its identifier for that column in the insert.
The PageSink, in finish, will return its identifier in the fragment slice passed back to the coordinator.
The coordinator will then insert all the successful page sink ID fragments from the successful PageSinks into an additional temporary helper table.
In the final insert into ... select from, we do a semi-join to that helper table created in step 7, thereby including only the rows inserted by successful tasks.

Non-technical explanation

This guarantees the ability to only include data from successful tasks. There is no additional necessary cleanup of data from failed tasks. This PR avoids changing any behavior of existing JDBC clients, including in 3rd party plugins, unless you opt-in by extending FaultTolerantJdbcClient. I have opted in for the PostgreSqlClient, and the MySqlClient.

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(X) Release notes are required, with the following suggested text:
PostgreSQL and MySQL connectors now fault tolerant and only include data from successful tasks, automatically discarding any data from retried tasks.

# Section
* Allow writes to PostgreSQL and MySQL connectors when fault tolerant execution is enabled (`retry-policy` is set to `TASK` or `QUERY`). (https://github.com/starburstdata/stargate/issues/3877)

cla-bot · 2022-10-03T20:12:43Z

Thank you for your pull request and welcome to the Trino community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. Continue to work with us on the review and improvements in this PR, and submit the signed CLA to [email protected]. Photos, scans, or digitally-signed PDF files are all suitable. Processing may take a few days. The CLA needs to be on file before we merge your changes. For more information, see https://github.com/trinodb/cla

findepi · 2022-10-04T14:58:14Z

Atomic writes for PostgreSQL and MySQL connectors

Writes appear atomic in these connectors, because they are publish in ConnectorMetadata.finish* methods.

Did you mean "Support fault tolerant execution when writing to PostgreSQL or MySQL connector" ?

findepi · 2022-10-04T14:58:24Z

cc @losipiuk @hashhar

mwd410 · 2022-10-04T18:42:57Z

@findepi yes - sorry for the confusion. my brain was pinned to the language of the original issue assigned to me, and even though it felt like the writes already were atomic, i was just going along with the language. i've adjusted the verbiage in the PR, though i now want to change the name of the AtomicJdbcClient to FaultTolerantJdbcClient and fix any other mention of atomic in my code.

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorPageSinkProvider.java

losipiuk · 2022-10-04T20:39:01Z

@mwd410 tests are failing. Please make sure there are no change-relevant tests, fixing those may make shape of PR significantly.

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/FaultTolerantJdbcClient.java

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/mapping/DefaultIdentifierMapping.java

core/trino-main/src/main/java/io/trino/execution/TaskId.java

losipiuk · 2022-10-21T09:13:58Z

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java

@@ -928,17 +928,9 @@ protected String createTableSql(RemoteTableName remoteTableName, List<String> co
    }

    @Override
-    protected String buildInsertSql(ConnectorSession session, RemoteTableName targetTable, RemoteTableName sourceTable, List<String> columnNames)
+    protected String postProcessInsertTableClause(ConnectorSession session)


pass source sql statement as arguemnt and return postporcessed one (with suffix appended)

losipiuk · 2022-10-21T09:17:30Z

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java

@@ -735,7 +743,7 @@ public void rollback()
    public ConnectorInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle, List<ColumnHandle> columns, RetryMode retryMode)
    {
        verify(!((JdbcTableHandle) tableHandle).isSynthetic(), "Not a table reference: %s", tableHandle);
-        if (retryMode != NO_RETRIES) {
+        if (retryMode != NO_RETRIES && !jdbcClient.supportsFaultTolerantExecution(session)) {


Maybe move this checks inside client itself. Now you are checking here, and there doing same check again inside to decide which code flow to follow - it is confusing.
Also please emit different error message if connector does not support retries at all and when connetor supports retries but NON_TRANSACTIONAL_INSERT is set. Currently we would get This connector does not support query retries for latter case and it is not a good error message.

Also then maybe we do not need to expose supportsFaultTolerantExecution() from JdbcClient

losipiuk · 2022-10-21T09:18:02Z

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java

@@ -698,7 +699,7 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle
    @Override
    public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional<ConnectorTableLayout> layout, RetryMode retryMode)
    {
-        if (retryMode != NO_RETRIES) {
+        if (retryMode != NO_RETRIES && !jdbcClient.supportsFaultTolerantExecution(session)) {


same comment as for beginInsert

losipiuk · 2022-10-21T09:37:54Z

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -844,16 +1013,25 @@ public void rollbackCreateTable(ConnectorSession session, JdbcOutputTableHandle
        }
    }

+    @Override
+    public boolean supportsFaultTolerantExecution(ConnectorSession session)


given internal uses of this method it should rather be "shouldUseFaultTolerantExecution"

losipiuk

Looks good. One comment really.

findepi · 2023-02-06T14:39:59Z

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

    public BaseJdbcClient(
            BaseJdbcConfig config,
            String identifierQuote,
            ConnectionFactory connectionFactory,
            QueryBuilder queryBuilder,
            IdentifierMapping identifierMapping)
+    {
+        this(


Deprecate the old constructor and remove all usages of it.

mwd410 changed the title ~~Atomic writes for PostgreSQL and MySQL connectors~~ Fault Tolerant Execution for PostgreSQL and MySQL connectors Oct 4, 2022

mwd410 force-pushed the mdeady-atomic-writes-3877 branch from 1961526 to 19486e8 Compare October 4, 2022 18:47

cla-bot bot added the cla-signed label Oct 4, 2022

mwd410 force-pushed the mdeady-atomic-writes-3877 branch from 19486e8 to 88b4988 Compare October 4, 2022 18:57

arhimondr self-requested a review October 4, 2022 19:11

findepi requested review from hashhar and findepi October 4, 2022 19:42

losipiuk self-requested a review October 4, 2022 20:25