feat(snowflake): add missing pushdown_deny_usernames config to be used when use_queries_v2

[sgomezvillamor]

The legacy SnowflakeQueriesSource inherits from SnowflakeQueriesExtractorConfig and so exposes the config param

datahub/metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_queries.py

Lines 113 to 115 in e2b1ed7

	class SnowflakeQueriesSourceConfig(
	SnowflakeQueriesExtractorConfig, SnowflakeIdentifierConfig, SnowflakeFilterConfig
	):

and then, uses it

datahub/metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_queries.py

Lines 330 to 338 in e2b1ed7

	def fetch_query_log(
	self, users: UsersMapping
	) -> Iterable[Union[PreparsedQuery, TableRename, TableSwap]]:
	query_log_query = _build_enriched_query_log_query(
	start_time=self.config.window.start_time,
	end_time=self.config.window.end_time,
	bucket_duration=self.config.window.bucket_duration,
	deny_usernames=self.config.pushdown_deny_usernames,
	)

However, the SnowflakeV2Source doesn't

datahub/metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_config.py

Lines 182 to 190 in e2b1ed7

	class SnowflakeV2Config(
	SnowflakeConfig,
	SnowflakeUsageConfig,
	StatefulLineageConfigMixin,
	StatefulUsageConfigMixin,
	StatefulProfilingConfigMixin,
	ClassificationSourceConfigMixin,
	IncrementalPropertiesConfigMixin,
	):

and so the proposal in this PR is to define the config param so it can be passed accordingly when instantiating the SnowflakeQueriesExtractor

Checklist

• The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
• Links to related issues (if applicable)
• Tests for the changes have been added/updated (if applicable)
• Docs related to the changes have been added/updated (if applicable). If a new feature has been added a Usage Guide has been added for the same.
• For any breaking change/potential downtime/deprecation/big changes an entry has been made in Updating DataHub

[codecov]

❌ 5 Tests Failed:

Tests completed	Failed	Passed	Skipped
1995	5	1990	40

View the top 3 failed tests by shortest run time

tests.integration.kafka-connect.test_kafka_connect::test_kafka_connect_ingest_stateful

Stack Traces | 0.001s run time

kafka_connect_runner = Services(_docker_compose=DockerComposeExecutor(_compose_command='docker compose', _compose_files=['.../runner/work/d...tegration/kafka-connect/docker-compose.override.yml'], _compose_project_name='pytest4874-kafka-connect'), _services={})

    @pytest.fixture(scope="module")
    def loaded_kafka_connect(kafka_connect_runner):
        # # Setup mongo cluster
        command = "docker exec test_mongo mongosh test_db -f /scripts/mongo-init.js"
        ret = subprocess.run(command, shell=True, capture_output=True)
        assert ret.returncode == 0
    
        # Creating MySQL source with no transformations , only topic prefix
        r = requests.post(
            KAFKA_CONNECT_ENDPOINT,
            headers={"Content-Type": "application/json"},
            data="""{
                "name": "mysql_source1",
                "config": {
                    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                    "mode": "incrementing",
                    "incrementing.column.name": "id",
                    "topic.prefix": "test-mysql-jdbc-",
                    "tasks.max": "1",
                    "connection.url": "${env:MYSQL_CONNECTION_URL}"
                }
            }
            """,
        )
>       assert r.status_code == 201  # Created
E       assert 500 == 201
E        +  where 500 = <Response [500]>.status_code

.../integration/kafka-connect/test_kafka_connect.py:111: AssertionError

tests.integration.kafka-connect.test_kafka_connect::test_kafka_connect_mongosourceconnect_ingest

Stack Traces | 0.001s run time

kafka_connect_runner = Services(_docker_compose=DockerComposeExecutor(_compose_command='docker compose', _compose_files=['.../runner/work/d...tegration/kafka-connect/docker-compose.override.yml'], _compose_project_name='pytest4874-kafka-connect'), _services={})

    @pytest.fixture(scope="module")
    def loaded_kafka_connect(kafka_connect_runner):
        # # Setup mongo cluster
        command = "docker exec test_mongo mongosh test_db -f /scripts/mongo-init.js"
        ret = subprocess.run(command, shell=True, capture_output=True)
        assert ret.returncode == 0
    
        # Creating MySQL source with no transformations , only topic prefix
        r = requests.post(
            KAFKA_CONNECT_ENDPOINT,
            headers={"Content-Type": "application/json"},
            data="""{
                "name": "mysql_source1",
                "config": {
                    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                    "mode": "incrementing",
                    "incrementing.column.name": "id",
                    "topic.prefix": "test-mysql-jdbc-",
                    "tasks.max": "1",
                    "connection.url": "${env:MYSQL_CONNECTION_URL}"
                }
            }
            """,
        )
>       assert r.status_code == 201  # Created
E       assert 500 == 201
E        +  where 500 = <Response [500]>.status_code

.../integration/kafka-connect/test_kafka_connect.py:111: AssertionError

tests.integration.kafka-connect.test_kafka_connect::test_kafka_connect_s3sink_ingest

Stack Traces | 0.001s run time

kafka_connect_runner = Services(_docker_compose=DockerComposeExecutor(_compose_command='docker compose', _compose_files=['.../runner/work/d...tegration/kafka-connect/docker-compose.override.yml'], _compose_project_name='pytest4874-kafka-connect'), _services={})

    @pytest.fixture(scope="module")
    def loaded_kafka_connect(kafka_connect_runner):
        # # Setup mongo cluster
        command = "docker exec test_mongo mongosh test_db -f /scripts/mongo-init.js"
        ret = subprocess.run(command, shell=True, capture_output=True)
        assert ret.returncode == 0
    
        # Creating MySQL source with no transformations , only topic prefix
        r = requests.post(
            KAFKA_CONNECT_ENDPOINT,
            headers={"Content-Type": "application/json"},
            data="""{
                "name": "mysql_source1",
                "config": {
                    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                    "mode": "incrementing",
                    "incrementing.column.name": "id",
                    "topic.prefix": "test-mysql-jdbc-",
                    "tasks.max": "1",
                    "connection.url": "${env:MYSQL_CONNECTION_URL}"
                }
            }
            """,
        )
>       assert r.status_code == 201  # Created
E       assert 500 == 201
E        +  where 500 = <Response [500]>.status_code

.../integration/kafka-connect/test_kafka_connect.py:111: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
📢 Thoughts on this report? Let us know!

[hsheth2]

we should think about how we can avoid needing to specify every config twice - not sure if there's a good way to do it though

[sgomezvillamor]

While everything is possible, those refactors will be easier when we remove the code base for queries v1.

[sgomezvillamor]

Some CI errors about kafka connect, so unrelated to this PR. Merging.