Skip to content
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

Destinations v2: Do not dedup raw table #31520

Merged
merged 15 commits into from
Oct 25, 2023
Merged

Destinations v2: Do not dedup raw table #31520

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
5caaf84
stuff
edgao Oct 17, 2023
c257f9e
Automated Commit - Formatting Changes
edgao Oct 17, 2023
1bebfd8
stuff
edgao Oct 17, 2023
a9697b8
more stuff
edgao Oct 17, 2023
0ae0e67
more fixes
edgao Oct 17, 2023
73c54e7
Automated Commit - Formatting Changes
edgao Oct 17, 2023
5db5b36
only filter on extracted_at in non-dedup mode
edgao Oct 17, 2023
55e5506
Merge branch 'master' into edgao/skip_raw_dedup
edgao Oct 18, 2023
f37a463
test cleanup
edgao Oct 18, 2023
ae2d132
test cleanup
edgao Oct 18, 2023
34fc49a
extracted_at filter in dedup mode
edgao Oct 18, 2023
55a4470
Merge branch 'master' into edgao/skip_raw_dedup
edgao Oct 18, 2023
3b60147
Merge branch 'master' into edgao/skip_raw_dedup
edgao Oct 24, 2023
fa9f214
Merge branch 'master' into edgao/skip_raw_dedup
edgao Oct 25, 2023
b0ca13a
version bumps
edgao Oct 25, 2023
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
21 changes: 10 additions & 11 deletions ...irbyte/integrations/base/destination/typing_deduping/BaseSqlGeneratorIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -659,13 +659,11 @@ public void incrementalDedupNoCursor() throws Exception {
final List<JsonNode> actualRawRecords = dumpRawTableRecords(streamId);
final List<JsonNode> actualFinalRecords = dumpFinalTableRecords(streamId, "");
verifyRecordCounts(
1,
2,
actualRawRecords,
1,
actualFinalRecords);
assertAll(
() -> assertEquals("bar", actualRawRecords.get(0).get("_airbyte_data").get("string").asText()),
() -> assertEquals("bar", actualFinalRecords.get(0).get(generator.buildColumnId("string").name()).asText()));
assertEquals("bar", actualFinalRecords.get(0).get(generator.buildColumnId("string").name()).asText());
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

there's no longer any reason to assert the raw records, so only assert the final record.

}

@Test
Expand Down Expand Up @@ -796,10 +794,9 @@ public void cdcComplexUpdate() throws Exception {
destinationHandler.execute(sql);

verifyRecordCounts(
// We keep the newest raw record per PK
7,
11,
dumpRawTableRecords(streamId),
5,
6,
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is a behavior change; see https://airbytehq-team.slack.com/archives/C03C4AVJWG4/p1697486280303179

dumpFinalTableRecords(streamId, ""));
}

Expand All @@ -824,11 +821,12 @@ public void testCdcOrdering_updateAfterDelete() throws Exception {
streamId,
BaseTypingDedupingTest.readRecords("sqlgenerator/cdcordering_updateafterdelete_inputrecords.jsonl"));

final String sql = generator.updateTable(cdcIncrementalDedupStream, "", Optional.empty());
final Optional<Instant> minTimestampForSync = destinationHandler.getMinTimestampForSync(cdcIncrementalAppendStream.id());
final String sql = generator.updateTable(cdcIncrementalDedupStream, "", minTimestampForSync);
destinationHandler.execute(sql);

verifyRecordCounts(
1,
2,
dumpRawTableRecords(streamId),
0,
dumpFinalTableRecords(streamId, ""));
Expand Down Expand Up @@ -861,11 +859,12 @@ public void testCdcOrdering_insertAfterDelete() throws Exception {
"",
BaseTypingDedupingTest.readRecords("sqlgenerator/cdcordering_insertafterdelete_inputrecords_final.jsonl"));

final String sql = generator.updateTable(cdcIncrementalDedupStream, "", Optional.empty());
final Optional<Instant> minTimestampForSync = destinationHandler.getMinTimestampForSync(cdcIncrementalAppendStream.id());
final String sql = generator.updateTable(cdcIncrementalDedupStream, "", minTimestampForSync);
destinationHandler.execute(sql);

verifyRecordCounts(
1,
2,
dumpRawTableRecords(streamId),
1,
dumpFinalTableRecords(streamId, ""));
Expand Down
51 changes: 26 additions & 25 deletions ...java/io/airbyte/integrations/base/destination/typing_deduping/BaseTypingDedupingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

package io.airbyte.integrations.base.destination.typing_deduping;

import static org.junit.jupiter.api.Assertions.assertAll;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.ImmutableMap;
Expand Down Expand Up @@ -226,7 +228,7 @@ public void fullRefreshOverwrite() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1);

Expand Down Expand Up @@ -261,7 +263,7 @@ public void fullRefreshAppend() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1);

Expand All @@ -270,7 +272,7 @@ public void fullRefreshAppend() throws Exception {

runSync(catalog, messages2);

final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_raw.jsonl");
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_final.jsonl");
verifySyncResult(expectedRawRecords2, expectedFinalRecords2);
}
Expand Down Expand Up @@ -300,7 +302,7 @@ public void incrementalAppend() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1);

Expand All @@ -309,7 +311,7 @@ public void incrementalAppend() throws Exception {

runSync(catalog, messages2);

final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_raw.jsonl");
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_final.jsonl");
verifySyncResult(expectedRawRecords2, expectedFinalRecords2);
}
Expand Down Expand Up @@ -337,7 +339,7 @@ public void incrementalDedup() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_dedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_dedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1);

Expand All @@ -346,7 +348,7 @@ public void incrementalDedup() throws Exception {

runSync(catalog, messages2);

final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_incremental_dedup_raw.jsonl");
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords2 = readRecords("dat/sync2_expectedrecords_incremental_dedup_final.jsonl");
verifySyncResult(expectedRawRecords2, expectedFinalRecords2);
}
Expand All @@ -372,7 +374,7 @@ public void incrementalDedupDefaultNamespace() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_dedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_dedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1, null, streamName);

Expand All @@ -381,7 +383,7 @@ public void incrementalDedupDefaultNamespace() throws Exception {

runSync(catalog, messages2);

final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_incremental_dedup_raw.jsonl");
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords2 = readRecords("dat/sync2_expectedrecords_incremental_dedup_final.jsonl");
verifySyncResult(expectedRawRecords2, expectedFinalRecords2, null, streamName);
}
Expand Down Expand Up @@ -424,7 +426,7 @@ public void testIncrementalSyncDropOneColumn() throws Exception {

runSync(catalog, messages1);

final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_raw.jsonl");
final List<JsonNode> expectedRawRecords1 = readRecords("dat/sync1_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords1 = readRecords("dat/sync1_expectedrecords_nondedup_final.jsonl");
verifySyncResult(expectedRawRecords1, expectedFinalRecords1);

Expand All @@ -437,7 +439,7 @@ public void testIncrementalSyncDropOneColumn() throws Exception {
runSync(catalog, messages2);

// The raw data is unaffected by the schema, but the final table should not have a `name` column.
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_raw.jsonl");
final List<JsonNode> expectedRawRecords2 = readRecords("dat/sync2_expectedrecords_raw.jsonl");
final List<JsonNode> expectedFinalRecords2 = readRecords("dat/sync2_expectedrecords_fullrefresh_append_final.jsonl").stream()
.peek(record -> ((ObjectNode) record).remove(getSqlGenerator().buildColumnId("name").name()))
.toList();
Expand Down Expand Up @@ -500,12 +502,12 @@ public void incrementalDedupIdenticalName() throws Exception {
runSync(catalog, messages1);

verifySyncResult(
readRecords("dat/sync1_expectedrecords_dedup_raw.jsonl"),
readRecords("dat/sync1_expectedrecords_raw.jsonl"),
readRecords("dat/sync1_expectedrecords_dedup_final.jsonl"),
namespace1,
streamName);
verifySyncResult(
readRecords("dat/sync1_expectedrecords_dedup_raw2.jsonl"),
readRecords("dat/sync1_expectedrecords_raw2.jsonl"),
readRecords("dat/sync1_expectedrecords_dedup_final2.jsonl"),
namespace2,
streamName);
Expand All @@ -518,12 +520,12 @@ public void incrementalDedupIdenticalName() throws Exception {
runSync(catalog, messages2);

verifySyncResult(
readRecords("dat/sync2_expectedrecords_incremental_dedup_raw.jsonl"),
readRecords("dat/sync2_expectedrecords_raw.jsonl"),
readRecords("dat/sync2_expectedrecords_incremental_dedup_final.jsonl"),
namespace1,
streamName);
verifySyncResult(
readRecords("dat/sync2_expectedrecords_incremental_dedup_raw2.jsonl"),
readRecords("dat/sync2_expectedrecords_raw2.jsonl"),
readRecords("dat/sync2_expectedrecords_incremental_dedup_final2.jsonl"),
namespace2,
streamName);
Expand Down Expand Up @@ -585,16 +587,15 @@ public void identicalNameSimultaneousSync() throws Exception {
// And this will dump sync2's entire stdout to our stdout
endSync(sync2);

verifySyncResult(
readRecords("dat/sync1_expectedrecords_dedup_raw.jsonl"),
readRecords("dat/sync1_expectedrecords_dedup_final.jsonl"),
namespace1,
streamName);
verifySyncResult(
readRecords("dat/sync1_expectedrecords_dedup_raw2.jsonl"),
readRecords("dat/sync1_expectedrecords_dedup_final2.jsonl"),
namespace2,
streamName);
// For simplicity, don't verify the raw table. Assume that if the final table is correct, then
// the raw data is correct. This is generally a safe assumption.
assertAll(
() -> DIFFER.diffFinalTableRecords(
readRecords("dat/sync1_expectedrecords_dedup_final.jsonl"),
dumpFinalTableRecords(namespace1, streamName)),
() -> DIFFER.diffFinalTableRecords(
readRecords("dat/sync1_expectedrecords_dedup_final2.jsonl"),
dumpFinalTableRecords(namespace2, streamName)));
}

@Test
Expand Down
2 changes: 1 addition & 1 deletion ...ase-typing-deduping-test/src/main/resources/sqlgenerator/cdcupdate_inputrecords_raw.jsonl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,5 @@
{"_airbyte_raw_id": "4d8674a5-eb6e-41ca-a310-69c64c88d101", "_airbyte_extracted_at": "2023-01-01T00:00:00Z", "_airbyte_data": {"id1": 0, "id2": 100, "updated_at": "2023-01-01T05:00:00Z", "_ab_cdc_deleted_at": null, "string": "zombie_returned"}}
// CDC generally outputs an explicit null for deleted_at, but verify that we can also handle the case where deleted_at is unset.
{"_airbyte_raw_id": "f0b59e49-8c74-4101-9f14-cb4d1193fd5a", "_airbyte_extracted_at": "2023-01-01T00:00:00Z", "_airbyte_data": {"id1": 4, "id2": 100, "updated_at": "2023-01-01T06:00:00Z", "string": "charlie"}}
// Verify that we can handle weird values in deleted_at
// Invalid values in _ab_cdc_deleted_at result in the record NOT being deleted. This behavior is up for debate, but it's an extreme edge case so not a high priority.
{"_airbyte_raw_id": "d4e1d989-c115-403c-9e68-5d320e6376bb", "_airbyte_extracted_at": "2023-01-01T00:00:00Z", "_airbyte_data": {"id1": 5, "id2": 100, "updated_at": "2023-01-01T07:00:00Z", "_ab_cdc_deleted_at": {}, "string": "david1"}}
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/destination-bigquery/metadata.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ data:
connectorSubtype: database
connectorType: destination
definitionId: 22f6c74f-5699-40ff-833c-4a879ea40133
dockerImageTag: 2.1.6
dockerImageTag: 2.2.0
dockerRepository: airbyte/destination-bigquery
githubIssueLabel: destination-bigquery
icon: bigquery.svg
Expand Down
Loading
Loading