Python source cdk does not support incremental sync with a cursor field configured by user

[marc-marketparts]

Enviroment

• Airbyte version: v0.30.16-alpha
• OS Version / Instance: Ubuntu 20.04.2 LTS on Windows 10 WSL 2
• Deployment: n/a
• Source Connector and version: python connector in development (Python source generator)
• Destination Connector and version: n/a
• Severity: Medium
• Step where error happened: Incremental sync with cursor field configured by the end user

Current Behavior

Incremental sync with a cursor field configured by the end user does not work with the current implementation of the python cdk.

There seems to be some flows in cursor field implementation in AbstractSource and Stream classes:

1. The method Stream::supports_incremental() rely on a source defined cursor_field (self.cursor_field) which is unknown in case of user configuration (and not set at configuration time). So the method will always return False in the latter because self.cursor_field will be left empty.
   

   airbyte/airbyte-integrations/bases/base-python/base_python/cdk/streams/core.py

   Lines 73 to 81 in 928aa62

   	@property
   	def supports_incremental(self) -> bool:
   	"""
   	:return: True if this stream supports incrementally reading data
   	"""
   	return len(self._wrapped_cursor_field()) > 0
   	def _wrapped_cursor_field(self) -> List[str]:
   	return [self.cursor_field] if isinstance(self.cursor_field, str) else self.cursor_field

2. If we override supports_incremental to return True, another issue occurs in the Stream::get_update_state() call. When a user select a cursor field in the UI, the configured cursor field is never passed to this method and therefore not accessible in it.
   When the initial state is empty, we cannot know which cursor field has been chosen, and so cannot generate the correct new state.
   

   airbyte/airbyte-integrations/bases/base-python/base_python/cdk/abstract_source.py

   Lines 135 to 144 in 928aa62

   	records = stream_instance.read_records(
   	sync_mode=SyncMode.incremental,
   	stream_slice=slice,
   	stream_state=stream_state,
   	cursor_field=configured_stream.cursor_field or None,
   	)
   	for record_data in records:
   	record_counter += 1
   	yield self._as_airbyte_record(stream_name, record_data)
   	stream_state = stream_instance.get_updated_state(stream_state, record_data)

This behavior works for source defined cursor only, as the cursor_field is accessible by referencing the property self.cursor_field defined in the source code of the stream.

Expected Behavior

The implementation of incremental sync must be out of the box, without any need to override python cdk core methods.

Logs

n/a

Steps to Reproduce

1. Implement a python source which inherits from AbstractSource
2. Implement an incremental stream with a user defined cursor field

Are you willing to submit a PR?

No

Here are some suggestions to correct this behaviour:

1. Add a condition on source_defined_cursor in Stream::supports_incremental():
   --> if False, we return True,
   --> if not, we return the result of _wrapped_cursor_field()

airbyte/airbyte-integrations/bases/base-python/base_python/cdk/streams/core.py

Lines 74 to 78 in 928aa62

	def supports_incremental(self) -> bool:
	"""
	:return: True if this stream supports incrementally reading data
	"""
	return len(self._wrapped_cursor_field()) > 0

• Add "cursor_field", to the method signature in the class Stream.get_update_state() as it is done on methods read_records() and stream_slices()
  

  airbyte/airbyte-integrations/bases/base-python/base_python/cdk/streams/core.py

  Line 150 in 928aa62

  def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]):

• Change the method call in AbstractSource::_read_incremental() to pass the "configured_stream.cursor_field"
  

  airbyte/airbyte-integrations/bases/base-python/base_python/cdk/abstract_source.py

  Line 144 in 928aa62

  stream_state = stream_instance.get_updated_state(stream_state, record_data)

NB: The two latter are not backward compatible.

[marcosmarxm]

@Phlair can you give your opinion here?

[Phlair]

This is a good spot. My observation aligns with the Current Behaviour in the issue description. Thanks for creating this issue @marc-marketparts!