Added new metadata field forced-replication-method

[atttiwari]

Description of change

SAC-28842
Added new metadata field in catalog's metadata, field name is forced-replication-method.

Manual QA steps

• Discovery: not tested
• Sync: not tested
• Unit Tests: not tested
• Integration test: not tested

Risks

• Tap may re-sync entire streams or lose incremental progress.

Rollback steps

• revert this branch

AI generated code

https://internal.qlik.dev/general/ways-of-working/code-reviews/#guidelines-for-ai-generated-code

• this PR has been written with the help of GitHub Copilot or another generative AI tool

[Copilot]

Pull Request Overview

This PR adds support for a new metadata field forced-replication-method to control replication strategy on a per-table basis. The implementation determines whether tables should use incremental or full table replication based on manifest configuration.

Key Changes:

• Added logic to extract and track incremental status from table manifests
• Implemented forced-replication-method metadata field with INCREMENTAL or FULL_TABLE values
• Refactored metadata handling to support the new field before converting to list format

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The incremental status is being overwritten on each iteration for the same table_name. If a table appears in multiple manifests with different incremental values, only the last one will be retained. Consider using a set or checking for conflicts to ensure consistent behavior across manifests.

Suggested change

	table_name_to_incremental = {}
	for all_table_manifests in manifests.values():
	for table_name, table_manifest in all_table_manifests.items():
	table_name_to_columns[table_name].update(set(table_manifest['columns']))
	table_name_to_incremental[table_name] = table_manifest.get('incremental', False)
	for table_name, columns in table_name_to_columns.items():
	schema = generate_fake_schema(columns)
	mdata = load_metadata(table_name, schema)
	if table_name_to_incremental.get(table_name, False):
	table_name_to_incremental = defaultdict(set)
	for all_table_manifests in manifests.values():
	for table_name, table_manifest in all_table_manifests.items():
	table_name_to_columns[table_name].update(set(table_manifest['columns']))
	table_name_to_incremental[table_name].add(table_manifest.get('incremental', False))
	# Resolve incremental status for each table, checking for conflicts
	resolved_table_name_to_incremental = {}
	for table_name, incremental_set in table_name_to_incremental.items():
	if len(incremental_set) > 1:
	raise ValueError(f"Conflicting incremental values for table '{table_name}': {incremental_set}")
	resolved_table_name_to_incremental[table_name] = next(iter(incremental_set))
	for table_name, columns in table_name_to_columns.items():
	schema = generate_fake_schema(columns)
	mdata = load_metadata(table_name, schema)
	if resolved_table_name_to_incremental.get(table_name, False):

[atttiwari]

addressed

[RushiT0122]

Suggested change

	if not flags:
	replication = 'FULL_TABLE'
	elif flags == {True}:
	replication = 'INCREMENTAL'
	elif flags == {False}:
	replication = 'FULL_TABLE'
	else:
	LOGGER.warning("Conflicting incremental flags for %s: %s. Defaulting to \
	INCREMENTAL.", table_name, flags)
	replication = 'INCREMENTAL'
	if flags == {True}:
	replication = 'INCREMENTAL'
	elif not flags or flags == {False}:
	replication = 'FULL_TABLE'
	else:
	LOGGER.warning("Conflicting incremental flags for %s: %s. Defaulting to \
	INCREMENTAL.", table_name, flags)
	replication = 'INCREMENTAL'