Merge ydb-platform#8604 (ydb-platform#8643)

ydb/core/fq/libs/db_id_async_resolver_impl/mdb_endpoint_generator.cpp

@@ -13,9 +13,8 @@ namespace NFq {
     constexpr ui32 CLICKHOUSE_HTTP_SECURE_PORT = 8443;
     constexpr ui32 CLICKHOUSE_HTTP_INSECURE_PORT = 8123;
 
-    // Managed PostgreSQL provides the only port both for secure and insecure connections
+    // Managed PostgreSQL and Greenplum provide the only port both for secure and insecure connections
     constexpr ui32 POSTGRESQL_PORT = 6432;
-
     constexpr ui32 GREENPLUM_PORT = 6432;
 
     constexpr ui32 MYSQL_PORT = 3306;

ydb/library/yql/providers/generic/connector/tests/utils/docker_compose.py

@@ -38,7 +38,7 @@ def __init__(self, docker_compose_yml_path: os.PathLike):
         self.docker_compose_yml_path = docker_compose_yml_path
 
         with open(self.docker_compose_yml_path) as f:
-            self.docker_compose_yml_data = yaml.load(f)
+            self.docker_compose_yml_data = yaml.load(f, Loader=yaml.FullLoader)
 
     def get_external_port(self, service_name: str, internal_port: int) -> int:
         cmd = [

ydb/tests/fq/generic/docker-compose.yml

@@ -18,6 +18,11 @@ services:
     image: ghcr.io/ydb-platform/fq-connector-go:v0.4.9@sha256:3a1fe086be50c0edbae2c2b284aee5ce76bd056d7f46cb460919ec37a6f8ab5c
     ports:
     - "2130"
+  greenplum:
+    container_name: tests-fq-generic-greenplum
+    image: ghcr.io/ydb-platform/fq-connector-go_greenplum:6.25.3-6432@sha256:9e862b05719b289b447562fbce6c003916a764a549f924a4175eecd7e7891a0b
+    volumes:
+    - ./greenplum/init_db.sh:/init_db.sh
   postgresql:
     command: -p 6432
     container_name: tests-fq-generic-postgresql

ydb/tests/fq/generic/greenplum/Dockerfile

@@ -0,0 +1,4 @@
+FROM ghcr.io/ydb-platform/fq-connector-go_greenplum:6.25.3@sha256:0627a657b179ff73949fec05201f3e164b92639281eff248cd07669ce7247eec
+
+# For the sake of simplicity of tests, we force Greenplum to use the same port that it uses within MDB
+RUN find /data -type f -exec sed -i 's/5432/6432/' "{}" +;

ydb/tests/fq/generic/greenplum/README.md

@@ -0,0 +1 @@
+Docker image built from this Dockerfile is pushed as `ghcr.io/ydb-platform/fq-connector-go_greenplum:6.25.3-6432@sha256:9e862b05719b289b447562fbce6c003916a764a549f924a4175eecd7e7891a0b`. No need to rebuild them every time.

ydb/tests/fq/generic/greenplum/init_db.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+set -e
+
+psql -p 6432 -v ON_ERROR_STOP=1 --username gpadmin --dbname template1 <<-EOSQL
+    CREATE TABLE simple_table (number INT);
+    INSERT INTO simple_table VALUES ((3)), ((14)), ((15));
+
+    CREATE TABLE join_table (id INT, data bytea);
+    INSERT INTO join_table VALUES (1, 'gp10'), (2, 'gp20'), (3, 'gp30');
+EOSQL

ydb/tests/fq/generic/test_greenplum.py

@@ -0,0 +1,47 @@
+import logging
+import pytest
+
+import ydb.public.api.protos.draft.fq_pb2 as fq
+import ydb.public.api.protos.ydb_value_pb2 as ydb
+from ydb.tests.tools.fq_runner.kikimr_utils import yq_v2
+
+from ydb.tests.tools.fq_runner.fq_client import FederatedQueryClient
+from ydb.tests.fq.generic.utils.settings import Settings
+
+
+class TestGreenplum:
+    @yq_v2
+    @pytest.mark.parametrize("fq_client", [{"folder_id": "my_folder"}], indirect=True)
+    def test_simple(self, fq_client: FederatedQueryClient, settings: Settings):
+        table_name = "simple_table"
+        conn_name = f"conn_{table_name}"
+        query_name = f"query_{table_name}"
+
+        fq_client.create_greenplum_connection(
+            name=conn_name,
+            database_name=settings.greenplum.dbname,
+            database_id="greenplum_cluster_id",
+            login=settings.greenplum.username,
+            password=settings.greenplum.password,
+        )
+
+        sql = Rf"""
+            SELECT *
+            FROM {conn_name}.{table_name} ORDER BY number;
+            """
+
+        query_id = fq_client.create_query(query_name, sql, type=fq.QueryContent.QueryType.ANALYTICS).result.query_id
+        fq_client.wait_query_status(query_id, fq.QueryMeta.COMPLETED)
+
+        data = fq_client.get_result_data(query_id)
+        result_set = data.result.result_set
+        logging.debug(str(result_set))
+        assert len(result_set.columns) == 1
+        assert result_set.columns[0].name == "number"
+        assert result_set.columns[0].type == ydb.Type(
+            optional_type=ydb.OptionalType(item=ydb.Type(type_id=ydb.Type.INT32))
+        )
+        assert len(result_set.rows) == 3
+        assert result_set.rows[0].items[0].int32_value == 3
+        assert result_set.rows[1].items[0].int32_value == 14
+        assert result_set.rows[2].items[0].int32_value == 15

ydb/tests/fq/generic/test_join.py

@@ -41,6 +41,7 @@ def test_simple(self, fq_client: FederatedQueryClient, settings: Settings, query
             database_id=settings.ydb.dbname,
         )
 
+        # FIXME: research why test starts failing if we add Greenplum
         sql = fR'''
             SELECT pg.data AS data_pg, ch.data AS data_ch, ydb.data AS data_ydb
             FROM {pg_conn_name}.{table_name} AS pg

ydb/tests/fq/generic/utils/endpoint_determiner.py

@@ -4,20 +4,26 @@
 import yatest.common
 
 
-# TODO: avoid duplication with ydb/library/yql/providers/generic/connector/tests/utils/docker_compose.py
 class EndpointDeterminer:
     docker_compose_bin: os.PathLike
     docker_compose_yml: os.PathLike
 
     def __init__(self, docker_compose_yml: os.PathLike):
-        self.docker_compose_bin = yatest.common.build_path('library/recipes/docker_compose/bin/docker-compose')
+        self.docker_compose_bin = yatest.common.build_path("library/recipes/docker_compose/bin/docker-compose")
         self.docker_compose_yml = docker_compose_yml
 
     def get_port(self, service_name: str, internal_port: int) -> int:
-        cmd = [self.docker_compose_bin, '-f', self.docker_compose_yml, 'port', service_name, str(internal_port)]
+        cmd = [
+            self.docker_compose_bin,
+            "-f",
+            self.docker_compose_yml,
+            "port",
+            service_name,
+            str(internal_port),
+        ]
         try:
             out = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
-            external_port = int(out.split(b':')[1])
+            external_port = int(out.split(b":")[1])
             return external_port
         except subprocess.CalledProcessError as e:
             raise RuntimeError(f"docker-compose error: {e.output} (code {e.returncode})")

ydb/tests/fq/generic/utils/settings.py

@@ -9,6 +9,8 @@
 
 @dataclass
 class Settings:
+    # infrastructure services
+
     @dataclass
     class Connector:
         grpc_host: str
@@ -29,6 +31,8 @@ class TokenAccessorMock:
 
     token_accessor_mock: TokenAccessorMock
 
+    # databases
+
     @dataclass
     class ClickHouse:
         dbname: str
@@ -38,6 +42,14 @@ class ClickHouse:
 
     clickhouse: ClickHouse
 
+    @dataclass
+    class Greenplum:
+        dbname: str
+        username: str
+        password: str
+
+    greenplum: Greenplum
+
     @dataclass
     class PostgreSQL:
         dbname: str
@@ -77,6 +89,11 @@ def from_env(cls) -> 'Settings':
                 password='password',
                 protocol='native',
             ),
+            greenplum=cls.Greenplum(
+                dbname='template1',
+                username='gpadmin',
+                password='123456',
+            ),
             postgresql=cls.PostgreSQL(
                 dbname='db',
                 username='user',

ydb/tests/fq/generic/ya.make

@@ -64,6 +64,7 @@ PEERDIR(
 TEST_SRCS(
     conftest.py
     test_clickhouse.py
+    test_greenplum.py
     test_join.py
     test_postgresql.py
     test_streaming_join.py

ydb/tests/tools/fq_runner/fq_client.py

@@ -421,22 +421,6 @@ def create_yds_connection(self, name, database=None, endpoint=None, database_id=
         request.content.acl.visibility = visibility
         return self.create_connection(request, check_issues)
 
-    @retry.retry_intrusive
-    def create_postgresql_connection(self, name, database_name, database_id, login, password,
-                                     secure=False, visibility=fq.Acl.Visibility.PRIVATE, auth_method=AuthMethod.service_account('sa'), check_issues=True):
-        request = fq.CreateConnectionRequest()
-        request.content.name = name
-        pg = request.content.setting.postgresql_cluster
-        pg.database_name = database_name
-        pg.database_id = database_id
-        pg.secure = secure
-        pg.login = login
-        pg.password = password
-
-        pg.auth.CopyFrom(auth_method)
-        request.content.acl.visibility = visibility
-        return self.create_connection(request, check_issues)
-
     @retry.retry_intrusive
     def create_clickhouse_connection(self, name, database_name, database_id, login, password,
                                      secure=False, visibility=fq.Acl.Visibility.PRIVATE, auth_method=AuthMethod.service_account('sa'), check_issues=True):
@@ -453,6 +437,37 @@ def create_clickhouse_connection(self, name, database_name, database_id, login,
         request.content.acl.visibility = visibility
         return self.create_connection(request, check_issues)
 
+    @retry.retry_intrusive
+    def create_greenplum_connection(self, name, database_name, database_id, login, password,
+                                    secure=False, visibility=fq.Acl.Visibility.PRIVATE, auth_method=AuthMethod.service_account('sa'), check_issues=True):
+        request = fq.CreateConnectionRequest()
+        request.content.name = name
+        gp = request.content.setting.greenplum_cluster
+        gp.database_name = database_name
+        gp.database_id = database_id
+        gp.login = login
+        gp.password = password
+
+        gp.auth.CopyFrom(auth_method)
+        request.content.acl.visibility = visibility
+        return self.create_connection(request, check_issues)
+
+    @retry.retry_intrusive
+    def create_postgresql_connection(self, name, database_name, database_id, login, password,
+                                     secure=False, visibility=fq.Acl.Visibility.PRIVATE, auth_method=AuthMethod.service_account('sa'), check_issues=True):
+        request = fq.CreateConnectionRequest()
+        request.content.name = name
+        pg = request.content.setting.postgresql_cluster
+        pg.database_name = database_name
+        pg.database_id = database_id
+        pg.secure = secure
+        pg.login = login
+        pg.password = password
+
+        pg.auth.CopyFrom(auth_method)
+        request.content.acl.visibility = visibility
+        return self.create_connection(request, check_issues)
+
     @retry.retry_intrusive
     def list_connections(self, visibility, name_substring=None, limit=100, check_issues=True, page_token=""):
         request = fq.ListConnectionsRequest()

ydb/tests/tools/fq_runner/kikimr_utils.py

@@ -271,8 +271,9 @@ def is_applicable(self, request):
 
     def apply_to_kikimr(self, request, kikimr):
         kikimr.control_plane.fq_config['common']['disable_ssl_for_generic_data_sources'] = True
-        kikimr.control_plane.fq_config['control_plane_storage']['available_connection'].append('POSTGRESQL_CLUSTER')
         kikimr.control_plane.fq_config['control_plane_storage']['available_connection'].append('CLICKHOUSE_CLUSTER')
+        kikimr.control_plane.fq_config['control_plane_storage']['available_connection'].append('GREENPLUM_CLUSTER')
+        kikimr.control_plane.fq_config['control_plane_storage']['available_connection'].append('POSTGRESQL_CLUSTER')
         kikimr.control_plane.fq_config['control_plane_storage']['available_connection'].append('YDB_DATABASE')
 
         generic = {

ydb/tests/tools/mdb_mock/__main__.py

@@ -53,10 +53,31 @@ async def postgresql_handler(request):
     return web.Response(body=json.dumps({}))
 
 
+async def greenplum_handler(request):
+    cluster_id = request.match_info['cluster_id']
+
+    if cluster_id == 'greenplum_cluster_id':
+        return web.Response(
+            body=json.dumps(
+                {
+                    "hosts": [
+                        {
+                            "name": "greenplum",
+                            "type": "MASTER",
+                            "health": "ALIVE",
+                        }
+                    ]
+                }
+            )
+        )
+    return web.Response(body=json.dumps({}))
+
+
 def serve(port: int):
     app = web.Application()
     app.add_routes([web.get('/managed-clickhouse/v1/clusters/{cluster_id}/hosts', clickhouse_handler)])
     app.add_routes([web.get('/managed-postgresql/v1/clusters/{cluster_id}/hosts', postgresql_handler)])
+    app.add_routes([web.get('/managed-greenplum/v1/clusters/{cluster_id}/master-hosts', greenplum_handler)])
     web.run_app(app, port=port)