Validate SqoopHook connection string and disable extra options from public hook methods

airflow/providers/apache/sqoop/CHANGELOG.rst

@@ -28,6 +28,17 @@
 Changelog
 ---------
 
+4.0.0
+.....
+
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+The ``extra_import_options`` parameter in the ``import_table`` & ``import_query`` methods
+and the ``extra_export_options`` in the ``export_table`` methods of the ``SqoopHook``
+are no longer accepted as arguments for those methods. These should instead be passed
+as ``extra_options`` while initializing the Hook or via ``extra_options`` parameter to the
+operator which instantiates the hook with those given ``extra_options`` dictionary.
 
 3.2.1
 .....

airflow/providers/apache/sqoop/hooks/sqoop.py

@@ -46,6 +46,9 @@ class SqoopHook(BaseHook):
     :param num_mappers: Number of map tasks to import in parallel.
     :param properties: Properties to set via the -D argument
     :param libjars: Optional Comma separated jar files to include in the classpath.
+    :param extra_options:  Extra import/export options to pass as dict.
+        If a key doesn't have a value, just pass an empty string to it.
+        Don't include prefix of -- for sqoop options.
     """
 
     conn_name_attr = "conn_id"
@@ -62,6 +65,7 @@ def __init__(
         hcatalog_table: str | None = None,
         properties: dict[str, Any] | None = None,
         libjars: str | None = None,
+        extra_options: dict[str, Any] | None = None,
     ) -> None:
         # No mutable types in the default parameters
         super().__init__()
@@ -79,6 +83,7 @@ def __init__(
         self.num_mappers = num_mappers
         self.properties = properties or {}
         self.sub_process_pid: int
+        self._extra_options = extra_options
         self.log.info("Using connection to: %s:%s/%s", self.conn.host, self.conn.port, self.conn.schema)
 
     def get_conn(self) -> Any:
@@ -113,9 +118,6 @@ def popen(self, cmd: list[str], **kwargs: Any) -> None:
                 raise AirflowException(f"Sqoop command failed: {masked_cmd}")
 
     def _prepare_command(self, export: bool = False) -> list[str]:
-        if "?" in self.conn.host:
-            raise ValueError("The sqoop connection host should not contain a '?' character")
-
         sqoop_cmd_type = "export" if export else "import"
         connection_cmd = ["sqoop", sqoop_cmd_type]
 
@@ -156,6 +158,8 @@ def _prepare_command(self, export: bool = False) -> list[str]:
                 connect_str += f"/{self.conn.schema}"
             else:
                 connect_str += f";databaseName={self.conn.schema}"
+        if "?" in connect_str:
+            raise ValueError("The sqoop connection string should not contain a '?' character")
         connection_cmd += ["--connect", connect_str]
 
         return connection_cmd
@@ -181,7 +185,6 @@ def _import_cmd(
         split_by: str | None,
         direct: bool | None,
         driver: Any,
-        extra_import_options: Any,
     ) -> list[str]:
 
         cmd = self._prepare_command(export=False)
@@ -203,8 +206,8 @@ def _import_cmd(
         if driver:
             cmd += ["--driver", driver]
 
-        if extra_import_options:
-            for key, value in extra_import_options.items():
+        if self._extra_options:
+            for key, value in self._extra_options.items():
                 cmd += [f"--{key}"]
                 if value:
                     cmd += [str(value)]
@@ -222,7 +225,6 @@ def import_table(
         where: str | None = None,
         direct: bool = False,
         driver: Any = None,
-        extra_import_options: dict[str, Any] | None = None,
         schema: str | None = None,
     ) -> Any:
         """Import table from remote location to target dir.
@@ -240,11 +242,8 @@ def import_table(
         :param where: WHERE clause to use during import
         :param direct: Use direct connector if exists for the database
         :param driver: Manually specify JDBC driver class to use
-        :param extra_import_options: Extra import options to pass as dict.
-            If a key doesn't have a value, just pass an empty string to it.
-            Don't include prefix of -- for sqoop options.
         """
-        cmd = self._import_cmd(target_dir, append, file_type, split_by, direct, driver, extra_import_options)
+        cmd = self._import_cmd(target_dir, append, file_type, split_by, direct, driver)
 
         cmd += ["--table", table]
 
@@ -266,7 +265,6 @@ def import_query(
         split_by: str | None = None,
         direct: bool | None = None,
         driver: Any | None = None,
-        extra_import_options: dict[str, Any] | None = None,
     ) -> Any:
         """Import a specific query from the rdbms to hdfs.
 
@@ -278,11 +276,8 @@ def import_query(
         :param split_by: Column of the table used to split work units
         :param direct: Use direct import fast path
         :param driver: Manually specify JDBC driver class to use
-        :param extra_import_options: Extra import options to pass as dict.
-            If a key doesn't have a value, just pass an empty string to it.
-            Don't include prefix of -- for sqoop options.
         """
-        cmd = self._import_cmd(target_dir, append, file_type, split_by, direct, driver, extra_import_options)
+        cmd = self._import_cmd(target_dir, append, file_type, split_by, direct, driver)
         cmd += ["--query", query]
 
         self.popen(cmd)
@@ -302,7 +297,6 @@ def _export_cmd(
         input_optionally_enclosed_by: str | None = None,
         batch: bool = False,
         relaxed_isolation: bool = False,
-        extra_export_options: dict[str, Any] | None = None,
         schema: str | None = None,
     ) -> list[str]:
 
@@ -344,8 +338,8 @@ def _export_cmd(
         if export_dir:
             cmd += ["--export-dir", export_dir]
 
-        if extra_export_options:
-            for key, value in extra_export_options.items():
+        if self._extra_options:
+            for key, value in self._extra_options.items():
                 cmd += [f"--{key}"]
                 if value:
                     cmd += [str(value)]
@@ -373,7 +367,6 @@ def export_table(
         input_optionally_enclosed_by: str | None = None,
         batch: bool = False,
         relaxed_isolation: bool = False,
-        extra_export_options: dict[str, Any] | None = None,
         schema: str | None = None,
     ) -> None:
         """Export Hive table to remote location.
@@ -399,9 +392,6 @@ def export_table(
         :param batch: Use batch mode for underlying statement execution
         :param relaxed_isolation: Transaction isolation to read uncommitted
             for the mappers
-        :param extra_export_options: Extra export options to pass as dict.
-            If a key doesn't have a value, just pass an empty string to it.
-            Don't include prefix of -- for sqoop options.
         """
         cmd = self._export_cmd(
             table,
@@ -417,7 +407,6 @@ def export_table(
             input_optionally_enclosed_by,
             batch,
             relaxed_isolation,
-            extra_export_options,
             schema,
         )
 

airflow/providers/apache/sqoop/operators/sqoop.py

@@ -76,10 +76,7 @@ class SqoopOperator(BaseOperator):
     :param create_hcatalog_table: Have sqoop create the hcatalog table passed
         in or not
     :param properties: additional JVM properties passed to sqoop
-    :param extra_import_options: Extra import options to pass as dict.
-        If a key doesn't have a value, just pass an empty string to it.
-        Don't include prefix of -- for sqoop options.
-    :param extra_export_options: Extra export options to pass as dict.
+    :param extra_options:  Extra import/export options to pass as dict to the SqoopHook.
         If a key doesn't have a value, just pass an empty string to it.
         Don't include prefix of -- for sqoop options.
     :param libjars: Optional Comma separated jar files to include in the classpath.
@@ -105,9 +102,8 @@ class SqoopOperator(BaseOperator):
         "input_lines_terminated_by",
         "input_optionally_enclosed_by",
         "properties",
-        "extra_import_options",
+        "extra_options",
         "driver",
-        "extra_export_options",
         "hcatalog_database",
         "hcatalog_table",
         "schema",
@@ -148,8 +144,7 @@ def __init__(
         hcatalog_database: str | None = None,
         hcatalog_table: str | None = None,
         create_hcatalog_table: bool = False,
-        extra_import_options: dict[str, Any] | None = None,
-        extra_export_options: dict[str, Any] | None = None,
+        extra_options: dict[str, Any] | None = None,
         schema: str | None = None,
         libjars: str | None = None,
         **kwargs: Any,
@@ -185,8 +180,7 @@ def __init__(
         self.hcatalog_table = hcatalog_table
         self.create_hcatalog_table = create_hcatalog_table
         self.properties = properties
-        self.extra_import_options = extra_import_options or {}
-        self.extra_export_options = extra_export_options or {}
+        self.extra_options = extra_options or {}
         self.hook: SqoopHook | None = None
         self.schema = schema
         self.libjars = libjars
@@ -211,16 +205,9 @@ def execute(self, context: Context) -> None:
                 input_optionally_enclosed_by=self.input_optionally_enclosed_by,
                 batch=self.batch,
                 relaxed_isolation=self.relaxed_isolation,
-                extra_export_options=self.extra_export_options,
                 schema=self.schema,
             )
         elif self.cmd_type == "import":
-            # add create hcatalog table to extra import options if option passed
-            # if new params are added to constructor can pass them in here
-            # so don't modify sqoop_hook for each param
-            if self.create_hcatalog_table:
-                self.extra_import_options["create-hcatalog-table"] = ""
-
             if self.table and self.query:
                 raise AirflowException("Cannot specify query and table together. Need to specify either or.")
 
@@ -235,7 +222,6 @@ def execute(self, context: Context) -> None:
                     where=self.where,
                     direct=self.direct,
                     driver=self.driver,
-                    extra_import_options=self.extra_import_options,
                     schema=self.schema,
                 )
             elif self.query:
@@ -247,7 +233,6 @@ def execute(self, context: Context) -> None:
                     split_by=self.split_by,
                     direct=self.direct,
                     driver=self.driver,
-                    extra_import_options=self.extra_import_options,
                 )
             else:
                 raise AirflowException("Provide query or table parameter to import using Sqoop")
@@ -261,6 +246,12 @@ def on_kill(self) -> None:
         os.killpg(os.getpgid(self.hook.sub_process_pid), signal.SIGTERM)
 
     def _get_hook(self) -> SqoopHook:
+        """Returns a SqoopHook instance."""
+        # Add `create-hcatalog-table` to extra options if option passed to operator in case of `import`
+        # command. Similarly, if new parameters are added to the operator, you can pass them to
+        # `extra_options` so that you don't need to modify `SqoopHook` for each new parameter.
+        if self.cmd_type == "import" and self.create_hcatalog_table:
+            self.extra_options["create-hcatalog-table"] = ""
         return SqoopHook(
             conn_id=self.conn_id,
             verbose=self.verbose,
@@ -269,4 +260,5 @@ def _get_hook(self) -> SqoopHook:
             hcatalog_table=self.hcatalog_table,
             properties=self.properties,
             libjars=self.libjars,
+            extra_options=self.extra_options,
         )

airflow/providers/apache/sqoop/provider.yaml

@@ -23,6 +23,7 @@ description: |
 
 suspended: false
 versions:
+  - 4.0.0
   - 3.2.1
   - 3.2.0
   - 3.1.1