The recipe (postgres.yml) is below.

source: 
  type: postgres
  config: 
    host_port: <http://192.168.xxx.xxx:xxxx|192.168.xxx.xxx:xxxx>
    database: xxxx
    username: xxxx
    password: xxxx

sink: 
  type: "datahub-rest"
  config: 
    server: "<http://localhost:8080>"

@miniature-tiger-96062 Thank you for your patience. All logs are now available.

I’m trying to ingest from postgresql, but the goal is actually to ingest from Vertica. In this case, it seems that the plugin for vertica is not implemented yet, so do I need to use the source type for other sql alchemy databases?

I have to use a driver for vertica because psycopg2 gives me an error. Can you tell me the best driver for vertica and how to incorporate that driver into the datahub-ingestion container?

We have a generic source for sql databases that can connect using sql alchemy. You may want to check this out - https://datahubproject.io/docs/metadata-ingestion/source_docs/sqlalchemy

I found “sqlalchemy-vertica-python” that use “v_catalog” instead of “pg_class”.

I installed “sqlalchemy-vertica-python” into “datahub-ingestion” container, but sqlalchemy use ‘sqlalchemy.dialects.postgresql’, not ‘vertica_python’.

So maybe try vertica+pyodbc?

It seems that sqlalchemy could load vertica+vertica_python, but it use the query for postgresql.

File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/case.py", line 1276, in _execute_context
    1186  def _execute_context(
    1187    self, dialect, constructor, statement, parameters, *args
    1188   ):
    (...)  
    1272        if fn(cursor, statement, parameters, context):
    1273          evt_handled = True
    1274          break
    1275      if not evt_handled:
--> 1276        self.dialect.do_execute(
    1277          cursor, statement, parameters, context
    ................................
    self = <sqlalchemy.engine.base.Connection object at 0x7fe8504dffd0>
    dialect = <sqlalchemy_vertica.dialect_vertica_python.VerticaDialect object at 0x7fe8504df820>
    constructor = <method 'DefaultExecutionContext._init_compiled' of <class 'sqlalchemy.dialects.postgresql.base.PGExecutionContext'> default.py:771>
    statement = "SELECT pg_get_view(c.oid) view_def FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = : schema AND c.relname = :view_name AND c.relkind IN ('v', 'm')"
    parameters = {'schema': 'public', 'view_name': 'V_FCT_ACCT_AGS'}
    args = (<sqlalchemy.dialects.postgresql.base.PGCompiler object at 0x7fe85041b1f0>, [{...}, ], )
    cursor = <vertica_python.vertica.cursor.Cursor object at 0x7fe8504875e0>
    context = <sqlalchemy.dialects.postgresql.base.PGExecutionContext object at 0x7fe85041beb0>
    evt_handled = False
    self.dialect.do_execute = <method 'DefaultDialect.do_execute' of <sqlalchemy_vertica.dialect_vertica_python.VerticaDialect object at 0x7fe8504df820> default.py:607>

---- (full traceback above) ----
File "/usr/local/lib/python3.8/site-packages/datahub/entrypoint.py", line 95, in main
    sys.exit(datahub(standalone_mode=False, **kwargs))
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1128, in __call__
    return self.main(*args, **kwargs)
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1053, in main
    rv = self.invoke(ctx)
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1395, in invoke
    return ctx.invoke(self.callback, **ctx.params)
File "/usr/local/lib/python3.8/site-packages/click/core.py", line 754, in invoke
    return __callback(*args, **kwargs)
File "/usr/local/lib/python3.8/site-packages/datahub/cli/ingest_cli.py", line 74, in run
    pipeline.run()
File "/usr/local/lib/python3.8/site-packages/datahub/ingestion/run/pipeline.py", line 148, in run
    for wu in itertools.islice(
File "/usr/local/lib/python3.8/site-packages/datahub/ingestion/source/sql/sql_common.py", line 353, in get_workunits
    yield from self.loop_views(inspector, schema, sql_config)
File "/usr/local/lib/python3.8/site-packages/datahub/ingestion/source/sql/sql_common.py", line 573, in loop_views
    view_definition = inspector.get_view_definition(view, schema)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/reflection.py", line 337, in get_view_definition
    return self.dialect.get_view_definition(
File "<string>", line 2, in get_view_definition
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/reflection.py", line 52, in cache
    ret = fn(self, con, *args, *kw)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/dialects/postgresql/base.py", line 3022, in get_view_definition
    view_def = connection.scalar(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 941, in scalar
    return self.execute(object_, *multiparams, **params).scalar()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
    return meth(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1124, in _execute_clauseelement
    ret = self._execute_context(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1316, in _execute_context
    self._handle_dbapi_exception(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1510, in _handle_dbapi_exception
    util.raise_(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
    self.dialect.do_execute(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
    cursor.execute(statement, parameters)
File "/usr/local/lib/python3.8/site-packages/vertica_python/vertica/cursor.py", line 222, in execute
    self._execute_simple_query(operation)
File "/usr/local/lib/python3.8/site-packages/vertica_python/vertica/cursor.py", line 606, in _execute_simple_query
    raise errors.QueryError.from_error_response(self._message, query)

ProgrammingError: (vertica_python.errors.MissingRelation) Severity: ERROR, Message: Relation "pg_class" does not exist, Sqlstate: 42V01, Routine: throwRelationDowsNotExist, File: /data/qb_workspaces/jenkins2/ReleaseBuilds/Hammerill/REL-10_0_1-x_hammermill/build/vertica/Catalog/CatalogLookup.cpp, Line: 3895, Error Code: 4566, SQL: "SELECT pg_get_viewdef(c.oid) view_def FROM pg_class c JOIN pg_namespace n ON n.oid - c.relnamespace WHERE n.nspname = 'xxxxx' AND c.relname = 'xxxxx' AND c.relkind IN ('v', 'm')"
[SQL: SELECT pg_get_viewdef(c.oid) view_def FROM pg_class c JOIN pg_namespace n ON n.oid - c.relnamespace WHERE n.nspname = 'xxxxx' AND c.relname = 'xxxxx' AND c.relkind IN ('v', 'm')]
[parameters: {'schema': 'xxxxx', 'view_name': 'xxxxx'}]

I think that it should be used below function in sqlalchemy-vertica-python for getting view information. https://github.com/bluelabsio/sqlalchemy-vertica-python/blob/99880f430ca4281eff845a47e424b33ff1520f12/sqla_vertica_python/vertica_python.py#L195 However, it seems that it is used below function in sqlalchemy/dialects/postgresql. https://github.com/sqlalchemy/sqlalchemy/blob/8d28bd8d4a0acc209884d29729020506a46f8bd4/lib/sqlalchemy/dialects/postgresql/base.py#L3584

When called ‘get_inspectors’ in ‘/datahub/ingestion/source/sql/sql_common.py’, the inspector is set to ‘postgresql’. https://github.com/linkedin/datahub/blob/adf9d2ead771fd2661645b0b9e5260f739a7a023/[…]tadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py It is assigned the inspector based on the url and config, but is it possible to specify the inspector explicitly in the recipe (yml)?

@miniature-tiger-96062 Should I submit it as an issue to github?