-
Notifications
You must be signed in to change notification settings - Fork 3.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
postgres-source: complete implementation for for ctid and xmin sync (#…
…27302) * initial ctid for testing * initial ctid for testing * initial ctid for testing * Automated Commit - Format and Process Resources Changes * add version and state type to xmin status * add logic to swtich between xmin and ctid sync * npe fixes * use enum * refactor * add relation node logic + validation for vacuuming + more refactor * refine test + make PR ready for review * remove un-wanted changes * missed this one * remove irrelevant comments * add more assertions * remove jdbc log * address review comments * 🤖 Auto format source-postgres code [skip ci] * skip streams under vacuum * 🤖 Auto format source-postgres code [skip ci] * update log message * 🤖 Auto format source-postgres code [skip ci] * comment * latest round of review comments * missed this file * 🤖 Auto format source-postgres code [skip ci] * source-postgres : Add logic to handle xmin wraparound (#27466) * Initial logic for xmin wraparound * Add tests * Address comments * add xmin-wraparound check * address review comments * 🤖 Auto format source-postgres code [skip ci] * missed this * 🤖 Auto format source-postgres code [skip ci] --------- Co-authored-by: Akash Kulkarni <akash@airbyte.io> Co-authored-by: Akash Kulkarni <113392464+akashkulk@users.noreply.github.com> Co-authored-by: octavia-squidington-iii <octavia-squidington-iii@users.noreply.github.com> --------- Co-authored-by: Rodi Reich Zilberman <867491+rodireich@users.noreply.github.com> Co-authored-by: rodireich <rodireich@users.noreply.github.com> Co-authored-by: octavia-squidington-iii <octavia-squidington-iii@users.noreply.github.com> Co-authored-by: Akash Kulkarni <akash@airbyte.io> Co-authored-by: Akash Kulkarni <113392464+akashkulk@users.noreply.github.com>
- Loading branch information
1 parent
3bd340a
commit 8c8f041
Showing
16 changed files
with
1,073 additions
and
74 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -57,3 +57,6 @@ jsonSchema2Pojo { | |
includeSetters = true | ||
} | ||
|
||
|
||
|
||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
...es/src/main/java/io/airbyte/integrations/source/postgres/ctid/AirbyteMessageWithCtid.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,19 @@ | ||
/* | ||
* Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.integrations.source.postgres.ctid; | ||
|
||
import io.airbyte.protocol.models.v0.AirbyteMessage; | ||
|
||
/** | ||
* ctid of rows is queried as part of our sync and is used to checkpoint to be able to restart | ||
* failed sync from a known last point. Since we never want to emit a ctid it is kept in a different | ||
* field, to save us an expensive JsonNode.remove() operation. | ||
* | ||
* @param recordMessage row fields to emit | ||
* @param ctid ctid | ||
*/ | ||
public record AirbyteMessageWithCtid(AirbyteMessage recordMessage, String ctid) { | ||
|
||
} |
46 changes: 46 additions & 0 deletions
46
.../main/java/io/airbyte/integrations/source/postgres/ctid/CtidPostgresSourceOperations.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,46 @@ | ||
/* | ||
* Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
*/ | ||
|
||
package io.airbyte.integrations.source.postgres.ctid; | ||
|
||
import com.fasterxml.jackson.databind.JsonNode; | ||
import com.fasterxml.jackson.databind.node.ObjectNode; | ||
import io.airbyte.commons.json.Jsons; | ||
import io.airbyte.integrations.source.postgres.PostgresSourceOperations; | ||
import java.sql.ResultSet; | ||
import java.sql.ResultSetMetaData; | ||
import java.sql.SQLException; | ||
import java.util.Collections; | ||
import java.util.Objects; | ||
|
||
public class CtidPostgresSourceOperations extends PostgresSourceOperations { | ||
|
||
private static final String CTID = "ctid"; | ||
|
||
public RowDataWithCtid recordWithCtid(final ResultSet queryContext) throws SQLException { | ||
// the first call communicates with the database. after that the result is cached. | ||
final ResultSetMetaData metadata = queryContext.getMetaData(); | ||
final int columnCount = metadata.getColumnCount(); | ||
final ObjectNode jsonNode = (ObjectNode) Jsons.jsonNode(Collections.emptyMap()); | ||
String ctid = null; | ||
for (int i = 1; i <= columnCount; i++) { | ||
final String columnName = metadata.getColumnName(i); | ||
if (columnName.equalsIgnoreCase(CTID)) { | ||
ctid = queryContext.getString(i); | ||
continue; | ||
} | ||
|
||
// convert to java types that will convert into reasonable json. | ||
copyToJsonField(queryContext, i, jsonNode); | ||
} | ||
|
||
assert Objects.nonNull(ctid); | ||
return new RowDataWithCtid(jsonNode, ctid); | ||
} | ||
|
||
public record RowDataWithCtid(JsonNode data, String ctid) { | ||
|
||
} | ||
|
||
} |
Oops, something went wrong.