Fixed MSSQL Source (#3877)

Fixed MSSQL Source (#3877)

catalog-rest-service/src/main/resources/json/schema/entity/services/connections/database/mssqlConnection.json

@@ -28,7 +28,7 @@
     "scheme": {
       "description": "SQLAlchemy driver scheme options.",
       "$ref": "#/definitions/mssqlScheme",
-      "default": "mssql+pyodbc"
+      "default": "mssql+pytds"
     },
     "username": {
       "description": "username to connect  to the MsSQL. This user should have privileges to read all the metadata in MsSQL.",
@@ -47,6 +47,10 @@
       "description": "Database of the data source. This is optional parameter, if you would like to restrict the metadata reading to a single database. When left blank , OpenMetadata Ingestion attempts to scan all the databases in MsSQL.",
       "type": "string"
     },
+    "uriString": {
+      "description": "Connection URI In case of pyodbc",
+      "type": "string"
+    },
     "connectionOptions": {
       "$ref": "connectionBasicType.json#/definitions/connectionOptions"
     },

ingestion-core/src/metadata/_version.py

@@ -7,5 +7,5 @@ Provides metadata version information.
 
 from incremental import Version
 
-__version__ = Version("metadata", 0, 9, 0, dev=10)
+__version__ = Version("metadata", 0, 9, 0, dev=11)
 __all__ = ["__version__"]

ingestion/examples/workflows/mssql.json

@@ -1,15 +1,20 @@
 {
   "source": {
     "type": "mssql",
+    "serviceName": "local_mssql",
+    "serviceConnection": {
       "config": {
-      "host_port": "localhost:1433",
+        "type": "MSSQL",
-      "service_name": "local_mssql",
         "database": "catalog_test",
-      "query": "select top 50 * from [{}].[{}]",
         "username": "sa",
         "password": "test!Password",
-      "table_filter_pattern": {
+        "hostPort": "localhost:1433"
-        "excludes": ["catalog_test.*"]
+      }
+    },
+    "sourceConfig": {
+      "config": {
+        "enableDataProfiler": false,
+        "sampleDataQuery": "select top 50 * from [{}].[{}]"
       }
     }
   },
@@ -17,11 +22,10 @@
     "type": "metadata-rest",
     "config": {}
   },
-  "metadata_server": {
+  "workflowConfig": { 
-    "type": "metadata-server",
+    "openMetadataServerConfig": {
-    "config": {
+      "hostPort": "http://localhost:8585/api",
-      "api_endpoint": "http://localhost:8585/api",
+      "authProvider": "no-auth"
-      "auth_provider_type": "no-auth"
     }
   }
 }

ingestion/src/metadata/ingestion/source/mssql.py

@@ -9,7 +9,6 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 """MSSQL source module"""
-from typing import Optional
 
 from metadata.generated.schema.entity.services.connections.database.mssqlConnection import (
     MssqlConnection,
@@ -17,25 +16,11 @@ from metadata.generated.schema.entity.services.connections.database.mssqlConnect
 from metadata.generated.schema.metadataIngestion.workflow import (
     OpenMetadataServerConfig,
 )
+from metadata.generated.schema.metadataIngestion.workflow import (
+    Source as WorkflowSource,
+)
+from metadata.ingestion.api.source import InvalidSourceException
 from metadata.ingestion.source.sql_source import SQLSource
-from metadata.ingestion.source.sql_source_common import SQLConnectionConfig
-
-
-class MssqlConfig(MssqlConnection, SQLConnectionConfig):
-    """MSSQL config -- extends SQLConnectionConfig class"""
-
-    use_pymssql: bool = False
-    use_pyodbc: bool = False
-    uri_string: str = ""
-    duration: Optional[int]
-
-    def get_connection_url(self):
-        if self.use_pyodbc:
-            self.scheme = self.scheme.mssql_pymssql
-            return f"{self.scheme}://{self.uri_string}"
-        if self.use_pymssql:
-            self.scheme = "mssql+pymssql"
-        return super().get_connection_url()
 
 
 class MssqlSource(SQLSource):
@@ -50,5 +35,10 @@ class MssqlSource(SQLSource):
     @classmethod
     def create(cls, config_dict, metadata_config: OpenMetadataServerConfig):
         """Create class instance"""
-        config = MssqlConfig.parse_obj(config_dict)
+        config: WorkflowSource = WorkflowSource.parse_obj(config_dict)
+        connection: MssqlConnection = config.serviceConnection.__root__.config
+        if not isinstance(connection, MssqlConnection):
+            raise InvalidSourceException(
+                f"Expected MssqlConnection, but got {connection}"
+            )
         return cls(config, metadata_config)

ingestion/src/metadata/ingestion/source/sql_source.py

@@ -217,6 +217,8 @@ class SQLSource(Source[OMetaDatabaseAndTable]):
                 ):
                     self.status.filter(schema, "Schema pattern not allowed")
                     continue
+                # Fetch tables by default
+                yield from self.fetch_tables(inspector, schema)
                 if self.source_config.includeViews:
                     yield from self.fetch_views(inspector, schema)
                 if self.source_config.markDeletedTables:

ingestion/src/metadata/utils/source_connections.py

@@ -14,6 +14,9 @@ Hosts the singledispatch to build source URLs
 from functools import singledispatch
 from urllib.parse import quote_plus
 
+from metadata.generated.schema.entity.services.connections.database.mssqlConnection import (
+    MssqlConnection,
+)
 from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
     MysqlConnection,
 )
@@ -22,16 +25,7 @@ from metadata.generated.schema.entity.services.connections.database.sqliteConnec
 )
 
 
-@singledispatch
+def get_connection_url_common(connection):
-def get_connection_url(connection):
-    raise NotImplemented(
-        f"Connection URL build not implemented for type {type(connection)}: {connection}"
-    )
-
-
-@get_connection_url.register
-def _(connection: MysqlConnection):
-
     url = f"{connection.scheme.value}://"
 
     if connection.username:
@@ -58,6 +52,25 @@ def _(connection: MysqlConnection):
     return url
 
 
+@singledispatch
+def get_connection_url(connection):
+    raise NotImplemented(
+        f"Connection URL build not implemented for type {type(connection)}: {connection}"
+    )
+
+
+@get_connection_url.register
+def _(connection: MysqlConnection):
+    return get_connection_url_common(connection)
+
+
+@get_connection_url.register
+def _(connection: MssqlConnection):
+    if connection.scheme.value == connection.scheme.mssql_pyodbc:
+        return f"{connection.scheme.value}://{connection.uriString}"
+    return get_connection_url_common(connection)
+
+
 @get_connection_url.register
 def _(connection: SQLiteConnection):
     """