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

fix(data-warehouse): Dont read last incremental value when the pipeline is being reset #27877

Merged
merged 2 commits into from
Jan 24, 2025
Merged

fix(data-warehouse): Dont read last incremental value when the pipeline is being reset #27877

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6c110df
Dont read last incremental value when the pipeline is being reset
Gilbert09 Jan 24, 2025
c63e52d
Log out sync type config
Gilbert09 Jan 24, 2025
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
29 changes: 17 additions & 12 deletions posthog/temporal/data_imports/workflow_activities/import_data_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,31 +94,36 @@ def import_data_activity_sync(inputs: ImportDataActivityInputs):
assert schema is not None
reset_pipeline = schema.sync_type_config.get("reset_pipeline", False) is True

logger.debug(f"schema.sync_type_config = {schema.sync_type_config}")
logger.debug(f"reset_pipeline = {reset_pipeline}")

schema = (
ExternalDataSchema.objects.prefetch_related("source")
.exclude(deleted=True)
.get(id=inputs.schema_id, team_id=inputs.team_id)
)

endpoints = [schema.name]
processed_incremental_last_value = None

if settings.TEMPORAL_TASK_QUEUE == DATA_WAREHOUSE_TASK_QUEUE_V2:
# Get the V2 last value, if it's not set yet (e.g. the first run), then fallback to the V1 value
processed_incremental_last_value = process_incremental_last_value(
schema.sync_type_config.get("incremental_field_last_value_v2"),
schema.sync_type_config.get("incremental_field_type"),
)
if reset_pipeline is not True:
if settings.TEMPORAL_TASK_QUEUE == DATA_WAREHOUSE_TASK_QUEUE_V2:
# Get the V2 last value, if it's not set yet (e.g. the first run), then fallback to the V1 value
processed_incremental_last_value = process_incremental_last_value(
schema.sync_type_config.get("incremental_field_last_value_v2"),
schema.sync_type_config.get("incremental_field_type"),
)

if processed_incremental_last_value is None:
if processed_incremental_last_value is None:
processed_incremental_last_value = process_incremental_last_value(
schema.sync_type_config.get("incremental_field_last_value"),
schema.sync_type_config.get("incremental_field_type"),
)
else:
processed_incremental_last_value = process_incremental_last_value(
schema.sync_type_config.get("incremental_field_last_value"),
schema.sync_type_config.get("incremental_field_type"),
)
else:
processed_incremental_last_value = process_incremental_last_value(
schema.sync_type_config.get("incremental_field_last_value"),
schema.sync_type_config.get("incremental_field_type"),
)

if schema.is_incremental:
logger.debug(f"Incremental last value being used is: {processed_incremental_last_value}")
Expand Down
Loading