PyAirbyte BigQuery Source Issue

[jroman-sh]

Airbyte Versions

airbyte==0.18.1
airbyte-api==0.49.4
airbyte-cdk==5.13.0
airbyte-protocol-models==0.5.1
airbyte_protocol_models_dataclasses==0.13.0
airbyte_protocol_models_pdv2==0.13.0

Problem Statement

I'm trying to sync data between BigQuery and a PostgreSQL instance, but I’m running into an issue that I believe is related to an Airbyte’s internal implementation problem.

Below is some mock code I’ve prepared to demonstrate my case:

from scripts.sync_cloud_sql.constants import BQ_CREDS, PG_CREDS

import airbyte as ab

if __name__ == "__main__":
    table_ref: str = "twinco-staging.playground.test"
    project_id, dataset_id, table_id = table_ref.split(".")

    source: ab.Source = ab.get_source(
        "source-bigquery",
        config={
            "project_id": project_id,
            "dataset_id": dataset_id,
            "credentials_json": BQ_CREDS,
        },
    )
    source.check()

    destination: ab.Destination = ab.get_destination("destination-postgres", config={**PG_CREDS, "schema": dataset_id})
    destination.check()

    result = destination.write(source, streams=[table_id], force_full_refresh=True, cache=False, state_cache=False)
    print(result)

This simple code syncs a table called twinco-staging.playground.test to PostgreSQL.
The table, which is a mock table, looks like the following:

When I run this simple sync job I get the following error.

AirbyteConnectorFailedError: Connector failed.
    Please review the log file for more information.
    Connector Name: 'source-bigquery'
    Exit Code: 1
    Log file: /tmp/airbyte/logs/source-bigquery/source-bigquery-log-JAA7EK146.log

Looking at the log:

...
2024-10-16 10:35:42 - INFO - ERROR i.a.c.i.b.AirbyteExceptionHandler(uncaughtException):64 Something went wrong in the connector. See the logs for more details. java.lang.NullPointerException: Cannot invoke "java.util.List.size()" because the return value of "io.airbyte.protocol.models.v0.ConfiguredAirbyteStream.getCursorField()" is null
	at io.airbyte.cdk.db.IncrementalUtils.getCursorField(IncrementalUtils.java:18) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.db.IncrementalUtils.getCursorFieldOptional(IncrementalUtils.java:29) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.source.relationaldb.AbstractDbSource.validateCursorFieldForIncrementalTables(AbstractDbSource.java:210) ~[airbyte-cdk-db-sources-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.source.relationaldb.AbstractDbSource.read(AbstractDbSource.java:167) ~[airbyte-cdk-db-sources-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.java:275) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:173) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.java:125) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.integrations.source.bigquery.BigQuerySource.main(BigQuerySource.java:220) ~[io.airbyte.airbyte-integrations.connectors-source-bigquery.jar:?]

2024-10-16 10:35:42 - ERROR - Something went wrong in the connector. See the logs for more details.
...

The error is because the air bytes tries to get the cursor of the table to get its size but the cursor is null. This happens at IncrementalUtils.kt:20

However, as I'm doing a full refresh and my source does not have any cursor field aribyte should not be looking at the cursor.

When debugging my code to get into the issue I looked at the configured_catalogues of my source:

ConfiguredAirbyteStream(
    stream=AirbyteStream(
        name='test', 
        json_schema={'type': 'object', 'properties': {'a': {'type': 'number', 'airbyte_type': 'integer'}, 'b': {'type': 'number', 'airbyte_type': 'integer'}, 'c': {'type': 'number', 'airbyte_type': 'integer'}}}, 
        supported_sync_modes=[<SyncMode.full_refresh: 'full_refresh'>, <SyncMode.incremental: 'incremental'>], 
        source_defined_cursor=None, 
        default_cursor_field=[], 
        source_defined_primary_key=[], 
        namespace='playground', 
        is_resumable=None),
    sync_mode=<SyncMode.incremental: 'incremental'>, 
    cursor_field=None, 
    destination_sync_mode=<DestinationSyncMode.overwrite: 'overwrite'>, 
    primary_key=[], 
    generation_id=None, 
    minimum_generation_id=None, 
    sync_id=None,
)

Here you can see that the cursor_field=None.
You may think why sync_mode=<SyncMode.incremental: 'incremental'>. This is configured by default by PyAirbyte's Source class and it's not possible to configure it otherwise.

Code Issue

Looking back at the first lines of the log we can see the following:

...
2024-10-16 10:35:42 - INFO - ERROR i.a.c.i.b.AirbyteExceptionHandler(uncaughtException):64 Something went wrong in the connector. See the logs for more details. java.lang.NullPointerException: Cannot invoke "java.util.List.size()" because the return value of "io.airbyte.protocol.models.v0.ConfiguredAirbyteStream.getCursorField()" is null
	at io.airbyte.cdk.db.IncrementalUtils.getCursorField(IncrementalUtils.java:18) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.db.IncrementalUtils.getCursorFieldOptional(IncrementalUtils.java:29) ~[airbyte-cdk-core-0.13.2.jar:?]
	at io.airbyte.cdk.integrations.source.relationaldb.AbstractDbSource.validateCursorFieldForIncrementalTables(AbstractDbSource.java:210) ~[airbyte-cdk-db-sources-0.13.2.jar:?]
...

The method validateCursorFieldForIncrementalTables calls getCursorFieldOptional which calls getCursorField where the NullPointerException error happens.

Looking at this method implementation we can see the following condition which I believe is wrong:
AbstractDBSource.kt:207

For each stream we get calculate hasSourceDefinedCursorwhich is a boolean variable that is true when the source has a cursor. As we saw before my source does not have a cursor.

Then we skip the validation if at least one of the following conditions is true:

• !tableNameToTable.containsKey(fullyQualifiedTableName): If the table name
• airbyteStream.syncMode != SyncMode.INCREMENTAL: If it's not incremental skip cursor validation
• ‼️ hasSourceDefinedCursor ‼️: This is the condition that does not make sense, if the source has a cursor we skip the validation and if it does not have it we continue with it calling getCursorFieldOptional. This does not make logical sense.

I believe that condition must be ! hasSourceDefinedCursor

if (
    !tableNameToTable.containsKey(fullyQualifiedTableName) ||
        airbyteStream.syncMode != SyncMode.INCREMENTAL ||
        !hasSourceDefinedCursor
) {
    continue
}

What do you think guys? If this is actually an issue can anyone help me write it?

Thanks