| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| import os |
| import sys |
| import tempfile |
| import types |
| import unittest |
| from unittest.mock import MagicMock, patch |
| |
| import pyarrow.fs as pafs |
| |
| from pypaimon.common.file_io import FileIO |
| from pypaimon.common.options import Options |
| from pypaimon.common.options.config import HdfsOptions |
| |
| |
| def _install_fake_hdfs_native(): |
| """Install a fake hdfs_native module (with .fsspec submodule) into |
| sys.modules. |
| |
| Returns (fake_module, mock_client_cls, mock_write_options_cls). |
| """ |
| fake = types.ModuleType("hdfs_native") |
| fake.Client = MagicMock(name="Client") |
| fake.WriteOptions = MagicMock(name="WriteOptions") |
| fsspec_mod = types.ModuleType("hdfs_native.fsspec") |
| fsspec_mod.HdfsFileSystem = MagicMock(name="HdfsFileSystem") |
| fsspec_mod.ViewfsFileSystem = MagicMock(name="ViewfsFileSystem") |
| fake.fsspec = fsspec_mod |
| sys.modules["hdfs_native"] = fake |
| sys.modules["hdfs_native.fsspec"] = fsspec_mod |
| return fake, fake.Client, fake.WriteOptions |
| |
| |
| def _uninstall_fake_hdfs_native(): |
| sys.modules.pop("hdfs_native", None) |
| sys.modules.pop("hdfs_native.fsspec", None) |
| # Also drop the cached HdfsNativeFileIO so a re-import sees the new fake |
| sys.modules.pop( |
| "pypaimon.filesystem.hdfs_native_file_io", None) |
| |
| |
| class HdfsOptionsTest(unittest.TestCase): |
| |
| def test_defaults(self): |
| opts = Options({}) |
| self.assertEqual(opts.get(HdfsOptions.HDFS_CLIENT_IMPL), "native") |
| self.assertTrue(opts.get(HdfsOptions.HDFS_CLIENT_FALLBACK_TO_PYARROW)) |
| self.assertIsNone(opts.get(HdfsOptions.HDFS_CONF_DIR)) |
| |
| def test_explicit_pyarrow(self): |
| opts = Options({"hdfs.client.impl": "pyarrow"}) |
| self.assertEqual(opts.get(HdfsOptions.HDFS_CLIENT_IMPL), "pyarrow") |
| |
| def test_explicit_fallback_false(self): |
| opts = Options({"hdfs.client.fallback-to-pyarrow": "false"}) |
| self.assertFalse(opts.get(HdfsOptions.HDFS_CLIENT_FALLBACK_TO_PYARROW)) |
| |
| |
| class HdfsNativeFileIORoutingTest(unittest.TestCase): |
| |
| def setUp(self): |
| _uninstall_fake_hdfs_native() |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def test_local_paths_unaffected(self): |
| fio = FileIO.get("file:///tmp/foo") |
| self.assertEqual(type(fio).__name__, "LocalFileIO") |
| |
| def test_default_hdfs_routes_to_native(self): |
| _install_fake_hdfs_native() |
| fio = FileIO.get("hdfs://ns/foo", Options({})) |
| self.assertEqual(type(fio).__name__, "HdfsNativeFileIO") |
| |
| def test_explicit_pyarrow_routes_to_pyarrow(self): |
| # No hdfs-native module needed; should go straight to pyarrow. |
| with patch( |
| "pypaimon.filesystem.pyarrow_file_io.PyArrowFileIO.__init__", |
| return_value=None, |
| ): |
| fio = FileIO.get( |
| "hdfs://ns/foo", |
| Options({"hdfs.client.impl": "pyarrow"}), |
| ) |
| self.assertEqual(type(fio).__name__, "PyArrowFileIO") |
| |
| def test_native_init_failure_falls_back_to_pyarrow(self): |
| # hdfs_native not installed; default fallback enabled. |
| _uninstall_fake_hdfs_native() |
| with patch( |
| "pypaimon.filesystem.pyarrow_file_io.PyArrowFileIO.__init__", |
| return_value=None, |
| ): |
| fio = FileIO.get("hdfs://ns/foo", Options({})) |
| self.assertEqual(type(fio).__name__, "PyArrowFileIO") |
| |
| def test_native_init_failure_no_fallback_raises(self): |
| _uninstall_fake_hdfs_native() |
| with self.assertRaises(ImportError): |
| FileIO.get( |
| "hdfs://ns/foo", |
| Options({"hdfs.client.fallback-to-pyarrow": "false"}), |
| ) |
| |
| def test_unsupported_impl_raises(self): |
| with self.assertRaises(ValueError) as ctx: |
| FileIO.get( |
| "hdfs://ns/foo", |
| Options({"hdfs.client.impl": "bogus"}), |
| ) |
| self.assertIn("Unsupported hdfs.client.impl", str(ctx.exception)) |
| |
| def test_env_var_override_when_option_absent(self): |
| _install_fake_hdfs_native() |
| with patch( |
| "pypaimon.filesystem.pyarrow_file_io.PyArrowFileIO.__init__", |
| return_value=None, |
| ): |
| with patch.dict(os.environ, {"PYPAIMON_HDFS_IMPL": "pyarrow"}): |
| fio = FileIO.get("hdfs://ns/foo", Options({})) |
| self.assertEqual(type(fio).__name__, "PyArrowFileIO") |
| |
| def test_option_wins_over_env_var(self): |
| _install_fake_hdfs_native() |
| with patch.dict(os.environ, {"PYPAIMON_HDFS_IMPL": "pyarrow"}): |
| fio = FileIO.get( |
| "hdfs://ns/foo", |
| Options({"hdfs.client.impl": "native"}), |
| ) |
| self.assertEqual(type(fio).__name__, "HdfsNativeFileIO") |
| |
| def test_viewfs_scheme_routes_to_native(self): |
| _install_fake_hdfs_native() |
| fio = FileIO.get("viewfs://cluster/foo", Options({})) |
| self.assertEqual(type(fio).__name__, "HdfsNativeFileIO") |
| |
| def test_empty_impl_option_treated_as_unset(self): |
| # Templated configs sometimes blank the option to opt out — that |
| # should fall through to the default ("native"), not raise |
| # "Unsupported hdfs.client.impl ''". |
| _install_fake_hdfs_native() |
| fio = FileIO.get("hdfs://ns/foo", Options({"hdfs.client.impl": ""})) |
| self.assertEqual(type(fio).__name__, "HdfsNativeFileIO") |
| |
| |
| class HdfsNativeFileIOInitTest(unittest.TestCase): |
| |
| def setUp(self): |
| self._fake, self._client_cls, self._wo_cls = _install_fake_hdfs_native() |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def _make(self, path, props=None): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| return HdfsNativeFileIO(path, Options(props or {})) |
| |
| def test_constructs_client_with_url(self): |
| self._make("hdfs://ns1/warehouse") |
| self._client_cls.assert_called_once() |
| _, kwargs = self._client_cls.call_args |
| self.assertEqual(kwargs.get("url"), "hdfs://ns1") |
| self.assertNotIn("config", kwargs) |
| |
| def test_viewfs_scheme_passes_through(self): |
| self._make("viewfs://cluster1/") |
| _, kwargs = self._client_cls.call_args |
| self.assertEqual(kwargs.get("url"), "viewfs://cluster1") |
| |
| def test_native_hadoop_keys_forwarded_as_config(self): |
| self._make("hdfs://ns1/foo", { |
| "dfs.nameservices": "ns1", |
| "dfs.ha.namenodes.ns1": "nn1,nn2", |
| "dfs.namenode.rpc-address.ns1.nn1": "host1:8020", |
| "fs.viewfs.mounttable.cluster.link./prod": "hdfs://ns1/prod", |
| "warehouse": "hdfs://ns1/warehouse", # should NOT be forwarded |
| }) |
| _, kwargs = self._client_cls.call_args |
| config = kwargs.get("config", {}) |
| self.assertEqual(config.get("dfs.nameservices"), "ns1") |
| self.assertEqual(config.get("dfs.ha.namenodes.ns1"), "nn1,nn2") |
| self.assertEqual( |
| config.get("dfs.namenode.rpc-address.ns1.nn1"), "host1:8020") |
| self.assertEqual( |
| config.get("fs.viewfs.mounttable.cluster.link./prod"), |
| "hdfs://ns1/prod") |
| self.assertNotIn("warehouse", config) |
| |
| def test_namespaced_overrides_forwarded(self): |
| self._make("hdfs://ns1/foo", { |
| "hdfs.config.dfs.client.read.shortcircuit": "true", |
| }) |
| _, kwargs = self._client_cls.call_args |
| config = kwargs.get("config", {}) |
| self.assertEqual( |
| config.get("dfs.client.read.shortcircuit"), "true") |
| |
| def test_conf_dir_from_option(self): |
| self._make("hdfs://ns1/foo", { |
| "hdfs.conf-dir": "/tmp/conf", |
| }) |
| _, kwargs = self._client_cls.call_args |
| self.assertEqual(kwargs.get("config_dir"), "/tmp/conf") |
| |
| def test_conf_dir_from_env(self): |
| env = dict(os.environ) |
| env["HADOOP_CONF_DIR"] = "/env/conf" |
| with patch.dict(os.environ, env, clear=True): |
| self._make("hdfs://ns1/foo") |
| _, kwargs = self._client_cls.call_args |
| self.assertEqual(kwargs.get("config_dir"), "/env/conf") |
| |
| def test_option_conf_dir_overrides_env(self): |
| env = dict(os.environ) |
| env["HADOOP_CONF_DIR"] = "/env/conf" |
| with patch.dict(os.environ, env, clear=True): |
| self._make("hdfs://ns1/foo", {"hdfs.conf-dir": "/opt/conf"}) |
| _, kwargs = self._client_cls.call_args |
| self.assertEqual(kwargs.get("config_dir"), "/opt/conf") |
| |
| @patch("pypaimon.filesystem._kerberos.subprocess.run") |
| def test_kerberos_principal_keytab_triggers_kinit(self, mock_kinit): |
| mock_kinit.return_value = MagicMock() |
| with tempfile.NamedTemporaryFile(suffix=".keytab") as keytab_file: |
| with patch.dict(os.environ, {"KRB5CCNAME": "/tmp/kc_test"}): |
| self._make("hdfs://ns1/foo", { |
| "security.kerberos.login.principal": "user@REALM", |
| "security.kerberos.login.keytab": keytab_file.name, |
| }) |
| kinit_calls = [ |
| c for c in mock_kinit.call_args_list |
| if c[0][0][0] == "kinit" |
| ] |
| self.assertEqual(len(kinit_calls), 1) |
| self.assertEqual( |
| kinit_calls[0][0][0], |
| ["kinit", "-kt", keytab_file.name, "user@REALM"], |
| ) |
| |
| def test_principal_without_keytab_raises(self): |
| with self.assertRaises(ValueError) as ctx: |
| self._make("hdfs://ns1/foo", { |
| "security.kerberos.login.principal": "user@REALM", |
| }) |
| self.assertIn("must be both set or both unset", str(ctx.exception)) |
| |
| @patch("pypaimon.filesystem._kerberos.subprocess.run") |
| def test_kerberos_preserves_FILE_prefix_on_krb5ccname(self, mock_kinit): |
| # If KRB5CCNAME had a `FILE:` qualifier, the rewrite after kinit |
| # must keep it so GSSAPI cache-type detection isn't perturbed. |
| mock_kinit.return_value = MagicMock() |
| with tempfile.NamedTemporaryFile(suffix=".keytab") as keytab_file: |
| with patch.dict(os.environ, |
| {"KRB5CCNAME": "FILE:/tmp/kc_test"}, |
| clear=True): |
| self._make("hdfs://ns1/foo", { |
| "security.kerberos.login.principal": "user@REALM", |
| "security.kerberos.login.keytab": keytab_file.name, |
| }) |
| self.assertEqual( |
| os.environ["KRB5CCNAME"], "FILE:/tmp/kc_test") |
| |
| @patch("pypaimon.filesystem._kerberos.get_ticket_cache_path", |
| return_value="/tmp/freshly_kinit_cache") |
| @patch("pypaimon.filesystem._kerberos.subprocess.run") |
| def test_kerberos_warns_when_overwriting_different_cache( |
| self, mock_kinit, _mock_cache, |
| ): |
| # Multi-principal in the same process clobbers KRB5CCNAME; we warn |
| # so the operator sees the race instead of silently mis-routing |
| # earlier instances' RPCs. |
| mock_kinit.return_value = MagicMock() |
| with tempfile.NamedTemporaryFile(suffix=".keytab") as keytab_file: |
| with patch.dict(os.environ, |
| {"KRB5CCNAME": "/tmp/some_other_cache"}, |
| clear=True): |
| with self.assertLogs( |
| "pypaimon.filesystem.hdfs_native_file_io", |
| level="WARNING", |
| ) as log_ctx: |
| self._make("hdfs://ns1/foo", { |
| "security.kerberos.login.principal": "user@REALM", |
| "security.kerberos.login.keytab": keytab_file.name, |
| }) |
| self.assertTrue( |
| any("Overwriting process-global KRB5CCNAME" in m |
| for m in log_ctx.output), |
| log_ctx.output, |
| ) |
| |
| @patch("pypaimon.filesystem._kerberos.get_ticket_cache_path", |
| return_value="/tmp/kc_test") |
| @patch("pypaimon.filesystem._kerberos.subprocess.run") |
| def test_kerberos_no_warning_when_cache_unchanged( |
| self, mock_kinit, _mock_cache, |
| ): |
| import logging as _logging |
| mock_kinit.return_value = MagicMock() |
| with tempfile.NamedTemporaryFile(suffix=".keytab") as keytab_file: |
| # Pre-existing value matches the kinit-resolved cache → no warn. |
| with patch.dict(os.environ, |
| {"KRB5CCNAME": "/tmp/kc_test"}, |
| clear=True): |
| # assertNoLogs is 3.10+; patch warning explicitly so older |
| # interpreters keep working too. |
| logger = _logging.getLogger( |
| "pypaimon.filesystem.hdfs_native_file_io") |
| with patch.object(logger, "warning") as warn: |
| self._make("hdfs://ns1/foo", { |
| "security.kerberos.login.principal": "user@REALM", |
| "security.kerberos.login.keytab": keytab_file.name, |
| }) |
| warn.assert_not_called() |
| |
| def test_unsupported_scheme_raises(self): |
| with self.assertRaises(ValueError): |
| self._make("s3://bucket/key") |
| |
| |
| class HdfsNativeFileIOOpsTest(unittest.TestCase): |
| |
| def setUp(self): |
| self._fake, self._client_cls, self._wo_cls = _install_fake_hdfs_native() |
| self._mock_client = MagicMock(name="ClientInstance") |
| self._client_cls.return_value = self._mock_client |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| self.fio = HdfsNativeFileIO("hdfs://ns/", Options({})) |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def _file_status(self, path, isdir=False, length=0, mtime=0): |
| s = MagicMock() |
| s.path = path |
| s.isdir = isdir |
| s.length = length |
| s.modification_time = mtime |
| return s |
| |
| def test_exists_true(self): |
| self._mock_client.get_file_info.return_value = self._file_status("/x") |
| self.assertTrue(self.fio.exists("/x")) |
| |
| def test_exists_false(self): |
| self._mock_client.get_file_info.side_effect = FileNotFoundError("nope") |
| self.assertFalse(self.fio.exists("/missing")) |
| |
| def test_get_file_status_adapts_to_pafs_filetype(self): |
| self._mock_client.get_file_info.return_value = self._file_status( |
| "/x", isdir=False, length=42, mtime=1700000000000, |
| ) |
| info = self.fio.get_file_status("/x") |
| self.assertEqual(info.path, "/x") |
| self.assertEqual(info.size, 42) |
| self.assertEqual(info.type, pafs.FileType.File) |
| self.assertIsNotNone(info.mtime) |
| |
| def test_list_status(self): |
| self._mock_client.list_status.return_value = iter([ |
| self._file_status("/x/a", isdir=False, length=10), |
| self._file_status("/x/b", isdir=True), |
| ]) |
| infos = self.fio.list_status("/x") |
| self.assertEqual(len(infos), 2) |
| self.assertEqual(infos[0].type, pafs.FileType.File) |
| self.assertEqual(infos[1].type, pafs.FileType.Directory) |
| self.assertIsNone(infos[1].size) |
| |
| def test_delete_missing_returns_false(self): |
| self._mock_client.get_file_info.side_effect = FileNotFoundError("nope") |
| self.assertFalse(self.fio.delete("/missing")) |
| self._mock_client.delete.assert_not_called() |
| |
| def test_delete_file(self): |
| self._mock_client.get_file_info.return_value = self._file_status("/x") |
| self._mock_client.delete.return_value = True |
| self.assertTrue(self.fio.delete("/x")) |
| self._mock_client.delete.assert_called_once_with("/x", False) |
| |
| def test_delete_nonempty_dir_without_recursive_raises(self): |
| self._mock_client.get_file_info.return_value = self._file_status( |
| "/x", isdir=True) |
| self._mock_client.list_status.return_value = iter([ |
| self._file_status("/x/a")]) |
| with self.assertRaises(OSError): |
| self.fio.delete("/x", recursive=False) |
| |
| def test_mkdirs_creates_when_missing(self): |
| self._mock_client.get_file_info.side_effect = FileNotFoundError("nope") |
| self.assertTrue(self.fio.mkdirs("/new")) |
| self._mock_client.mkdirs.assert_called_once_with( |
| "/new", create_parent=True) |
| |
| def test_mkdirs_idempotent_on_existing_dir(self): |
| self._mock_client.get_file_info.return_value = self._file_status( |
| "/x", isdir=True) |
| self.assertTrue(self.fio.mkdirs("/x")) |
| self._mock_client.mkdirs.assert_not_called() |
| |
| def test_mkdirs_existing_file_raises(self): |
| self._mock_client.get_file_info.return_value = self._file_status( |
| "/x", isdir=False) |
| with self.assertRaises(FileExistsError): |
| self.fio.mkdirs("/x") |
| |
| |
| class HdfsNativeAdaptersTest(unittest.TestCase): |
| |
| def setUp(self): |
| _install_fake_hdfs_native() |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def test_writer_adapter_tracks_position_and_closes_once(self): |
| from pypaimon.filesystem.hdfs_native_file_io import _HdfsWriterAdapter |
| fw = MagicMock() |
| fw.write.side_effect = lambda buf: len(buf) |
| adapter = _HdfsWriterAdapter(fw) |
| adapter.write(b"abc") |
| adapter.write(b"defg") |
| self.assertEqual(adapter.tell(), 7) |
| adapter.close() |
| adapter.close() # idempotent |
| fw.close.assert_called_once() |
| |
| def test_reader_adapter_seek_and_read(self): |
| from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter |
| fr = MagicMock() |
| fr.tell.side_effect = [20, 30] |
| fr.read.return_value = b"x" * 10 |
| adapter = _HdfsReaderAdapter(fr) |
| self.assertEqual(adapter.seek(20), 20) |
| fr.seek.assert_called_once_with(20, 0) |
| data = adapter.read(10) |
| self.assertEqual(data, b"x" * 10) |
| fr.read.assert_called_once_with(10) |
| self.assertEqual(adapter.tell(), 30) |
| |
| def test_reader_adapter_read_negative_reads_all(self): |
| from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter |
| fr = MagicMock() |
| fr.read.return_value = b"all-content" |
| adapter = _HdfsReaderAdapter(fr) |
| self.assertEqual(adapter.read(), b"all-content") |
| fr.read.assert_called_once_with(-1) |
| |
| def test_reader_adapter_close_releases_underlying(self): |
| from pypaimon.filesystem.hdfs_native_file_io import _HdfsReaderAdapter |
| fr = MagicMock() |
| adapter = _HdfsReaderAdapter(fr) |
| adapter.close() |
| adapter.close() # idempotent |
| fr.close.assert_called_once() |
| self.assertTrue(adapter.closed) |
| |
| |
| def _write_hadoop_xml(path, entries): |
| """Write a minimal Hadoop-style xml file at `path`.""" |
| body = ['<?xml version="1.0"?>', "<configuration>"] |
| for name, value in entries.items(): |
| body.append( |
| f" <property><name>{name}</name><value>{value}</value></property>" |
| ) |
| body.append("</configuration>") |
| with open(path, "w") as f: |
| f.write("\n".join(body)) |
| |
| |
| class ViewFsFallbackTest(unittest.TestCase): |
| """Cover _load_hadoop_xml + _maybe_inject_viewfs_fallback polyfill.""" |
| |
| def setUp(self): |
| _install_fake_hdfs_native() |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| self.Fio = HdfsNativeFileIO |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def test_load_hadoop_xml_merges_two_files(self): |
| with tempfile.TemporaryDirectory() as d: |
| _write_hadoop_xml(os.path.join(d, "core-site.xml"), |
| {"fs.defaultFS": "viewfs://c1"}) |
| _write_hadoop_xml(os.path.join(d, "hdfs-site.xml"), |
| {"dfs.nameservices": "ns1"}) |
| cfg = self.Fio._load_hadoop_xml(d) |
| self.assertEqual(cfg.get("fs.defaultFS"), "viewfs://c1") |
| self.assertEqual(cfg.get("dfs.nameservices"), "ns1") |
| |
| def test_load_hadoop_xml_missing_dir_returns_empty(self): |
| self.assertEqual(self.Fio._load_hadoop_xml(None), {}) |
| self.assertEqual(self.Fio._load_hadoop_xml("/no/such/dir/xyz"), {}) |
| |
| def test_load_hadoop_xml_malformed_file_skipped(self): |
| with tempfile.TemporaryDirectory() as d: |
| with open(os.path.join(d, "core-site.xml"), "w") as f: |
| f.write("<not really xml") |
| _write_hadoop_xml(os.path.join(d, "hdfs-site.xml"), |
| {"dfs.nameservices": "ns1"}) |
| cfg = self.Fio._load_hadoop_xml(d) |
| self.assertEqual(cfg, {"dfs.nameservices": "ns1"}) |
| |
| def test_fallback_from_existing_link_target(self): |
| overrides = {} |
| xml = { |
| "fs.viewfs.mounttable.c1.link./home": "hdfs://ns-prod/home", |
| "fs.viewfs.mounttable.c1.link./tmp": "hdfs://ns-tmp/tmp", |
| } |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual( |
| overrides.get("fs.viewfs.mounttable.c1.linkFallback"), |
| "hdfs://ns-prod/", |
| ) |
| |
| def test_fallback_from_nameservices_when_no_links(self): |
| overrides = {} |
| xml = {"dfs.nameservices": "nsA,nsB"} |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual( |
| overrides.get("fs.viewfs.mounttable.c1.linkFallback"), |
| "hdfs://nsA/", |
| ) |
| |
| def test_fallback_from_link_in_overrides_only(self): |
| # Zero-file viewfs setup: link.* arrives via catalog options |
| # (overrides), no hadoop xml present. The fallback must still be |
| # derived from the merged view. |
| overrides = { |
| "fs.viewfs.mounttable.c1.link./home": "hdfs://ns-prod/home", |
| } |
| xml = {} |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual( |
| overrides.get("fs.viewfs.mounttable.c1.linkFallback"), |
| "hdfs://ns-prod/", |
| ) |
| |
| def test_fallback_from_nameservices_in_overrides_only(self): |
| overrides = {"dfs.nameservices": "nsA,nsB"} |
| xml = {} |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual( |
| overrides.get("fs.viewfs.mounttable.c1.linkFallback"), |
| "hdfs://nsA/", |
| ) |
| |
| def test_fallback_already_in_xml_not_overridden(self): |
| overrides = {} |
| xml = { |
| "fs.viewfs.mounttable.c1.linkFallback": "hdfs://ns-existing/", |
| "dfs.nameservices": "nsA", |
| } |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertNotIn("fs.viewfs.mounttable.c1.linkFallback", overrides) |
| |
| def test_fallback_already_in_overrides_kept(self): |
| overrides = {"fs.viewfs.mounttable.c1.linkFallback": "hdfs://manual/"} |
| xml = {"dfs.nameservices": "nsA"} |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual( |
| overrides["fs.viewfs.mounttable.c1.linkFallback"], |
| "hdfs://manual/", |
| ) |
| |
| def test_hdfs_scheme_does_not_inject(self): |
| overrides = {} |
| xml = {"dfs.nameservices": "nsA"} |
| self.Fio._maybe_inject_viewfs_fallback("hdfs", "ns1", overrides, xml) |
| self.assertEqual(overrides, {}) |
| |
| def test_no_signal_no_inject(self): |
| overrides = {} |
| xml = {"unrelated.key": "v"} |
| self.Fio._maybe_inject_viewfs_fallback("viewfs", "c1", overrides, xml) |
| self.assertEqual(overrides, {}) |
| |
| def test_init_auto_injects_for_viewfs_uri(self): |
| with tempfile.TemporaryDirectory() as d: |
| _write_hadoop_xml( |
| os.path.join(d, "hdfs-site.xml"), |
| { |
| "dfs.nameservices": "ns1", |
| "fs.viewfs.mounttable.hadoop-lt-cluster.link./home": |
| "hdfs://ns1/home", |
| }, |
| ) |
| opts = Options({"hdfs.conf-dir": d}) |
| self.Fio("viewfs://hadoop-lt-cluster/home/x", opts) |
| client_cls = sys.modules["hdfs_native"].Client |
| _, kwargs = client_cls.call_args |
| config = kwargs.get("config", {}) |
| self.assertEqual( |
| config.get("fs.viewfs.mounttable.hadoop-lt-cluster.linkFallback"), |
| "hdfs://ns1/", |
| ) |
| |
| |
| class ToFilesystemPathTest(unittest.TestCase): |
| """Cover URI -> absolute-path normalisation for hdfs-native.""" |
| |
| def setUp(self): |
| _install_fake_hdfs_native() |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def _make(self, root): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| return HdfsNativeFileIO(root, Options({})) |
| |
| def test_viewfs_uri_same_cluster_returns_path(self): |
| fio = self._make("viewfs://cluster1/") |
| self.assertEqual( |
| fio.to_filesystem_path("viewfs://cluster1/home/hudi/x"), |
| "/home/hudi/x", |
| ) |
| |
| def test_viewfs_uri_no_path_returns_root(self): |
| fio = self._make("viewfs://cluster1/") |
| self.assertEqual(fio.to_filesystem_path("viewfs://cluster1"), "/") |
| |
| def test_viewfs_absolute_path_unchanged(self): |
| fio = self._make("viewfs://cluster1/") |
| self.assertEqual(fio.to_filesystem_path("/foo/bar"), "/foo/bar") |
| |
| def test_hdfs_uri_same_ns_returns_path(self): |
| fio = self._make("hdfs://ns1/") |
| self.assertEqual( |
| fio.to_filesystem_path("hdfs://ns1/foo/bar"), |
| "/foo/bar", |
| ) |
| |
| def test_hdfs_uri_different_ns_unchanged(self): |
| fio = self._make("hdfs://ns1/") |
| self.assertEqual( |
| fio.to_filesystem_path("hdfs://nsX/foo"), |
| "hdfs://nsX/foo", |
| ) |
| |
| def test_hdfs_client_with_viewfs_uri_unchanged(self): |
| fio = self._make("hdfs://ns1/") |
| # Different scheme; let hdfs-native error rather than silently rewrite. |
| self.assertEqual( |
| fio.to_filesystem_path("viewfs://cluster1/foo"), |
| "viewfs://cluster1/foo", |
| ) |
| |
| def test_exists_passes_path_only_to_client(self): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| fio = HdfsNativeFileIO("viewfs://cluster1/", Options({})) |
| client = sys.modules["hdfs_native"].Client.return_value |
| client.get_file_info.return_value = MagicMock( |
| path="/home/hudi/x", isdir=False, length=0, modification_time=0) |
| fio.exists("viewfs://cluster1/home/hudi/x") |
| client.get_file_info.assert_called_once_with("/home/hudi/x") |
| |
| |
| class FilesystemPropertyTest(unittest.TestCase): |
| """Cover the lazy pyarrow.fs facade backed by hdfs_native.fsspec.""" |
| |
| def setUp(self): |
| _install_fake_hdfs_native() |
| self._patcher = patch("pyarrow.fs.PyFileSystem") |
| self._handler_patcher = patch("pyarrow.fs.FSSpecHandler") |
| self.MockPyFs = self._patcher.start() |
| self.MockHandler = self._handler_patcher.start() |
| |
| def tearDown(self): |
| self._patcher.stop() |
| self._handler_patcher.stop() |
| _uninstall_fake_hdfs_native() |
| |
| def _make(self, root, props=None, xml_entries=None): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| if xml_entries: |
| d = tempfile.mkdtemp() |
| self.addCleanup(lambda: __import__("shutil").rmtree(d, ignore_errors=True)) |
| _write_hadoop_xml(os.path.join(d, "hdfs-site.xml"), xml_entries) |
| base_props = {"hdfs.conf-dir": d} |
| base_props.update(props or {}) |
| props = base_props |
| return HdfsNativeFileIO(root, Options(props or {})) |
| |
| def test_viewfs_uses_viewfs_fsspec_class(self): |
| fio = self._make("viewfs://cluster1/") |
| fs_instance = fio.filesystem # trigger lazy |
| VFs = sys.modules["hdfs_native.fsspec"].ViewfsFileSystem |
| HFs = sys.modules["hdfs_native.fsspec"].HdfsFileSystem |
| VFs.assert_called_once() |
| HFs.assert_not_called() |
| _, kwargs = VFs.call_args |
| self.assertEqual(kwargs.get("host"), "cluster1") |
| self.assertIs(fs_instance, self.MockPyFs.return_value) |
| |
| def test_hdfs_uses_hdfs_fsspec_class(self): |
| self._make("hdfs://ns1/").filesystem |
| HFs = sys.modules["hdfs_native.fsspec"].HdfsFileSystem |
| VFs = sys.modules["hdfs_native.fsspec"].ViewfsFileSystem |
| HFs.assert_called_once() |
| VFs.assert_not_called() |
| _, kwargs = HFs.call_args |
| self.assertEqual(kwargs.get("host"), "ns1") |
| |
| def test_lazy_caches_after_first_access(self): |
| fio = self._make("hdfs://ns1/") |
| first = fio.filesystem |
| second = fio.filesystem |
| self.assertIs(first, second) |
| HFs = sys.modules["hdfs_native.fsspec"].HdfsFileSystem |
| self.assertEqual(HFs.call_count, 1) |
| |
| def test_xml_and_catalog_options_merged_into_fsspec_storage_options(self): |
| fio = self._make( |
| "hdfs://ns1/", |
| props={"dfs.client.read.shortcircuit": "true"}, |
| xml_entries={"dfs.nameservices": "ns1"}, |
| ) |
| fio.filesystem # trigger |
| HFs = sys.modules["hdfs_native.fsspec"].HdfsFileSystem |
| _, kwargs = HFs.call_args |
| # Both xml and option keys should land in the fsspec kwargs. |
| self.assertEqual(kwargs.get("dfs.nameservices"), "ns1") |
| self.assertEqual(kwargs.get("dfs.client.read.shortcircuit"), "true") |
| |
| def test_catalog_option_overrides_xml(self): |
| fio = self._make( |
| "hdfs://ns1/", |
| props={"dfs.foo": "v_user"}, |
| xml_entries={"dfs.foo": "v_xml"}, |
| ) |
| fio.filesystem |
| HFs = sys.modules["hdfs_native.fsspec"].HdfsFileSystem |
| _, kwargs = HFs.call_args |
| self.assertEqual(kwargs.get("dfs.foo"), "v_user") |
| |
| def test_missing_fsspec_raises_clear_error(self): |
| fio = self._make("hdfs://ns1/") |
| # Remove the fsspec submodule but keep hdfs_native itself, to |
| # simulate an old/partial install. |
| sys.modules.pop("hdfs_native.fsspec", None) |
| sys.modules["hdfs_native"].fsspec = None |
| with self.assertRaises(RuntimeError) as ctx: |
| fio.filesystem |
| self.assertIn("hdfs-native fsspec adapter", str(ctx.exception)) |
| |
| |
| class PickleTest(unittest.TestCase): |
| """Cover __reduce__ so Ray / multiprocessing can ship FileIO.""" |
| |
| def setUp(self): |
| _install_fake_hdfs_native() |
| # Isolate from any HADOOP_CONF_DIR on the host so __reduce__'s |
| # env-derived config_dir pinning is deterministic across machines. |
| self._env_patcher = patch.dict(os.environ, {}, clear=True) |
| self._env_patcher.start() |
| |
| def tearDown(self): |
| self._env_patcher.stop() |
| _uninstall_fake_hdfs_native() |
| |
| def _make(self, path, props=None): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| return HdfsNativeFileIO(path, Options(props or {})) |
| |
| def test_reduce_returns_class_and_args(self): |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| fio = self._make("viewfs://cluster1/some/sub/path", |
| {"dfs.nameservices": "ns1"}) |
| cls, args = fio.__reduce__() |
| self.assertIs(cls, HdfsNativeFileIO) |
| path, options = args |
| # Path is rebuilt from scheme+netloc (path segment dropped) — that |
| # is intentional because __init__ ignores path beyond scheme+netloc. |
| self.assertEqual(path, "viewfs://cluster1") |
| self.assertEqual(options.to_map(), {"dfs.nameservices": "ns1"}) |
| |
| def test_reduce_for_empty_netloc(self): |
| fio = self._make("hdfs://") |
| _, (path, _) = fio.__reduce__() |
| self.assertEqual(path, "hdfs://") |
| |
| def test_reduce_pins_env_resolved_config_dir_into_options(self): |
| # config_dir resolved from $HADOOP_CONF_DIR should be carried into |
| # the pickled options so a worker on a host with a different env |
| # value still uses the driver's resolved directory. |
| with tempfile.TemporaryDirectory() as d: |
| with patch.dict(os.environ, {"HADOOP_CONF_DIR": d}, clear=True): |
| fio = self._make("hdfs://ns1/foo") |
| _, (_, options) = fio.__reduce__() |
| self.assertEqual(options.to_map().get("hdfs.conf-dir"), d) |
| |
| def test_reduce_does_not_override_explicit_conf_dir_option(self): |
| with tempfile.TemporaryDirectory() as opt_dir: |
| with patch.dict(os.environ, |
| {"HADOOP_CONF_DIR": "/env/dir"}, clear=True): |
| fio = self._make("hdfs://ns1/foo", |
| {"hdfs.conf-dir": opt_dir}) |
| _, (_, options) = fio.__reduce__() |
| self.assertEqual(options.to_map().get("hdfs.conf-dir"), opt_dir) |
| |
| def test_pickle_roundtrip_preserves_type_and_options(self): |
| import pickle |
| fio = self._make("hdfs://ns1/foo", |
| {"dfs.foo": "bar", "fs.viewfs.x": "y"}) |
| client_cls = sys.modules["hdfs_native"].Client |
| client_cls.reset_mock() |
| # Roundtrip via the highest pickle protocol. |
| blob = pickle.dumps(fio, protocol=pickle.HIGHEST_PROTOCOL) |
| restored = pickle.loads(blob) |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| self.assertIsInstance(restored, HdfsNativeFileIO) |
| self.assertEqual(restored.properties.to_map(), |
| {"dfs.foo": "bar", "fs.viewfs.x": "y"}) |
| # The original __init__ ran once; the unpickle ran __init__ again. |
| self.assertEqual(client_cls.call_count, 1) |
| |
| def test_pickle_with_viewfs_scheme(self): |
| import pickle |
| fio = self._make("viewfs://cluster1/") |
| restored = pickle.loads(pickle.dumps(fio)) |
| self.assertEqual(restored._scheme, "viewfs") |
| self.assertEqual(restored._netloc, "cluster1") |
| |
| def test_pickle_does_not_serialise_live_client(self): |
| # If the live _client were pickled, the call would fail (MagicMocks |
| # are picklable but the real RawClient would not be). This test |
| # documents the contract: __reduce__ MUST sidestep _client. |
| import pickle |
| fio = self._make("hdfs://ns1/") |
| blob = pickle.dumps(fio) |
| # The pickled blob should reference the constructor inputs only; |
| # specifically it should not embed the literal mock _client. |
| self.assertNotIn(b"_client", blob) |
| |
| |
| class HdfsNativeWriteFormatTest(unittest.TestCase): |
| """HdfsNativeFileIO is the default hdfs:// backend, so it must keep the |
| same write surface as the pyarrow backend it replaces — including |
| lance/vortex, which otherwise fall through to FileIO's NotImplementedError. |
| """ |
| |
| def setUp(self): |
| self._fake, self._client_cls, _ = _install_fake_hdfs_native() |
| self._client_cls.return_value = MagicMock(name="ClientInstance") |
| from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO |
| self.fio = HdfsNativeFileIO("hdfs://ns/", Options({})) |
| |
| def tearDown(self): |
| _uninstall_fake_hdfs_native() |
| |
| def test_lance_and_vortex_are_overridden(self): |
| # The regression this guards: with these unimplemented, an HDFS table |
| # using file.format=lance/vortex would hit FileIO's NotImplementedError. |
| from pypaimon.common.file_io import FileIO |
| self.assertIsNot( |
| type(self.fio).write_lance, FileIO.write_lance) |
| self.assertIsNot( |
| type(self.fio).write_vortex, FileIO.write_vortex) |
| |
| def test_write_lance_delegates_to_lance_specified(self): |
| import pyarrow |
| table = pyarrow.table({"a": [1, 2]}) |
| writer = MagicMock(name="LanceFileWriter") |
| fake_lance = types.ModuleType("lance") |
| fake_lance.file = types.SimpleNamespace( |
| LanceFileWriter=MagicMock(return_value=writer)) |
| with patch.dict(sys.modules, {"lance": fake_lance}), \ |
| patch("pypaimon.read.reader.lance_utils.to_lance_specified", |
| return_value=("hdfs://ns/x.lance", {"opt": "v"})) as spec: |
| self.fio.write_lance("hdfs://ns/x.lance", table) |
| spec.assert_called_once() |
| _, kwargs = fake_lance.file.LanceFileWriter.call_args |
| self.assertEqual(kwargs.get("storage_options"), {"opt": "v"}) |
| writer.close.assert_called_once() |
| |
| def test_write_vortex_delegates_to_vortex_specified(self): |
| import pyarrow |
| table = pyarrow.table({"a": [1, 2]}) |
| fake_vortex = types.ModuleType("vortex") |
| fake_vortex.array = MagicMock(return_value="varr") |
| fake_vortex.store = types.SimpleNamespace(from_url=MagicMock()) |
| fake_io = types.ModuleType("vortex._lib.io") |
| fake_io.write = MagicMock() |
| fake_lib = types.ModuleType("vortex._lib") |
| fake_lib.io = fake_io |
| fake_modules = { |
| "vortex": fake_vortex, |
| "vortex._lib": fake_lib, |
| "vortex._lib.io": fake_io, |
| } |
| with patch.dict(sys.modules, fake_modules), \ |
| patch("pypaimon.read.reader.vortex_utils.to_vortex_specified", |
| return_value=("hdfs://ns/x.vortex", None)): |
| self.fio.write_vortex("hdfs://ns/x.vortex", table) |
| fake_io.write.assert_called_once_with("varr", "hdfs://ns/x.vortex") |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
