This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6488-1366b872c70ab29924fa5a8470a1c2317f0eba4a in repository https://gitbox.apache.org/repos/asf/texera.git
commit aefae7b3757b079acef0c8254ac04afb06de4729 Author: Xinyuan Lin <[email protected]> AuthorDate: Fri Jul 17 17:55:11 2026 -0700 fix(workflow-core): run Iceberg local storage on Windows without winutils (#6488) ### What changes were proposed in this PR? On a Windows dev machine without a native Hadoop installation, every workflow execution fails at execution start when the `postgres` or `hadoop` Iceberg catalog type is configured. Hadoop's local file system applies POSIX permissions on **every** file/directory creation by shelling out to `%HADOOP_HOME%\bin\winutils.exe`, with no configuration flag to disable it — and Iceberg's `HadoopFileIO` (the local warehouse used by the `postgres` and `hadoop` catalog types) writes through exactly that path. The stock default catalog is `rest` (#6049), which writes through `S3FileIO` and is unaffected. This is a regression from the CVE-driven hadoop-common bumps: Windows dev worked on 3.3.1; after #6201 (3.4.3) and #6227 (3.5.0), 3.4+ applies POSIX permissions on every local file/directory creation, which pulls winutils in: ``` ExecutionStatsService (create runtime-stats Iceberg table) └─ IcebergUtil.createTable └─ JdbcTableOperations.doCommit → HadoopFileIO └─ RawLocalFileSystem.mkdirs / create └─ RawLocalFileSystem.setPermission ← on every file/dir creation └─ Shell.getSetPermissionCommand ← requires winutils.exe chmod └─ FileNotFoundException: Hadoop bin directory does not exist ``` POSIX permission bits carry no meaning on NTFS, so this PR skips them there instead of failing: | File | Change | |---|---| | `WinutilsFreeLocalFileSystem.scala` (new) | A `LocalFileSystem` whose `setPermission`/`setOwner` skip the winutils shell-out. Self-gating: when winutils **is** installed, or on non-Windows hosts, it delegates to Hadoop's default behavior. | | `IcebergUtil.scala` | New `newLocalHadoopConf()` used by `createHadoopCatalog`/`createPostgresCatalog`. Only when `Shell.WINDOWS && !Shell.hasWinutilsPath()` does it select the new file system via `fs.file.impl` (and disable the scheme-keyed `FileSystem` cache so an earlier cached instance cannot bypass the override). On all other hosts it returns the default `Configuration` unchanged. | Before → after: | Environment | Before | After | |---|---|---| | Windows + `postgres`/`hadoop` catalog, no winutils | any workflow run fails immediately | works | | Windows with winutils installed | works | unchanged (gate off, shim delegates) | | Linux / macOS / CI | works | unchanged (gate off, shim delegates) | | `rest` catalog (stock default) | unaffected (`S3FileIO`) | unaffected | ### Any related issues, documentation, discussions? Closes #6487 Related: #6150 (possible removal of the `hadoop` catalog type — the fix targets `HadoopFileIO`, which the `postgres` catalog also uses, so it is needed independently; the regression test currently drives `HadoopFileIO` through `createHadoopCatalog` as a lightweight vehicle and can be re-anchored on the local file system directly if the `hadoop` type is removed) ### How was this PR tested? - New regression test in `IcebergUtilSpec` (create + load a table via `createHadoopCatalog` in a temp local warehouse). Written first against stock `main` on a Windows machine without winutils: it fails with the exact production error (`RawLocalFileSystem.setPermission → Shell.getSetPermissionCommand → FileNotFoundException: Hadoop bin directory does not exist`), and passes with this fix. On Linux CI it exercises the unchanged default path. - `newLocalHadoopConf`'s winutils gate is injectable (defaults to the host's real winutils availability), so both branches — including the Windows-only override — are unit-tested on every platform: selection of `WinutilsFreeLocalFileSystem` via `fs.file.impl`, and the untouched-`Configuration` path. - New `WinutilsFreeLocalFileSystemSpec`: selection through `fs.file.impl` (same mechanism production uses), `mkdirs`/`create`/read/delete round-trip, and `setPermission`/`setOwner` no-throw. - Manually verified on Windows: workflow execution against the `postgres` catalog previously failed at `ExecutionStatsService` table creation and now succeeds end-to-end (`JdbcCatalog` + `HadoopFileIO` → local warehouse write). - `WorkflowCore/scalafmtCheckAll` and `WorkflowCore/scalafixAll --check` pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Fable 5) --- .../org/apache/texera/amber/util/IcebergUtil.scala | 36 +++++++++- .../amber/util/WinutilsFreeLocalFileSystem.scala | 51 +++++++++++++ .../apache/texera/amber/util/IcebergUtilSpec.scala | 40 +++++++++++ .../util/WinutilsFreeLocalFileSystemSpec.scala | 84 ++++++++++++++++++++++ 4 files changed, 208 insertions(+), 3 deletions(-) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/IcebergUtil.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/IcebergUtil.scala index b8e46e5cc5..4485a8cfbc 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/IcebergUtil.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/IcebergUtil.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.util import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, LargeBinary, Schema, Tuple} import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.util.Shell import org.apache.iceberg.catalog.{Catalog, SupportsNamespaces, TableIdentifier} import org.apache.iceberg.data.parquet.GenericParquetReaders import org.apache.iceberg.data.{GenericRecord, Record} @@ -58,10 +59,37 @@ object IcebergUtil { // Unique suffix for LARGE_BINARY field encoding private val LARGE_BINARY_FIELD_SUFFIX = "__texera_large_binary_ptr" + /** + * Creates the Hadoop `Configuration` used by catalogs that access the local file + * system through `HadoopFileIO`. + * + * On Windows hosts without a native Hadoop installation (winutils.exe), Hadoop's + * default local file system fails on every write because it shells out to winutils + * for chmod. In that case, swap in [[WinutilsFreeLocalFileSystem]], which skips + * permission operations, so local development works without installing winutils. + * On all other hosts the default configuration is returned unchanged. + * + * @param useWinutilsFreeLocalFs whether to swap in the winutils-free file system; + * defaults to the host's actual winutils availability + * and is only overridden in tests. + */ + private[util] def newLocalHadoopConf( + useWinutilsFreeLocalFs: Boolean = Shell.WINDOWS && !Shell.hasWinutilsPath() + ): Configuration = { + val conf = new Configuration() + if (useWinutilsFreeLocalFs) { + conf.set("fs.file.impl", classOf[WinutilsFreeLocalFileSystem].getName) + // The FileSystem cache is keyed by scheme (not by impl class), so a plain + // LocalFileSystem cached earlier by other code would bypass this override. + conf.setBoolean("fs.file.impl.disable.cache", true) + } + conf + } + /** * Creates and initializes a HadoopCatalog with the given parameters. - * - Uses an empty Hadoop `Configuration`, meaning the local file system (or `file:/`) will be used by default - * instead of HDFS. + * - Uses the Hadoop `Configuration` from [[newLocalHadoopConf]], meaning the local + * file system (or `file:/`) will be used by default instead of HDFS. * - The `warehouse` parameter specifies the root directory for storing table data. * - Sets the file I/O implementation to `HadoopFileIO`. * @@ -74,7 +102,7 @@ object IcebergUtil { warehouse: Path ): HadoopCatalog = { val catalog = new HadoopCatalog() - catalog.setConf(new Configuration) // Empty configuration, defaults to `file:/` + catalog.setConf(newLocalHadoopConf()) // Defaults to `file:/`, no HDFS catalog.initialize( catalogName, Map( @@ -128,6 +156,8 @@ object IcebergUtil { // Explicitly load the JDBC driver to avoid flaky CI failures. Class.forName("org.postgresql.Driver") val catalog = new JdbcCatalog() + // Must be set before initialize() so HadoopFileIO picks up this configuration. + catalog.setConf(newLocalHadoopConf()) catalog.initialize( catalogName, Map( diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystem.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystem.scala new file mode 100644 index 0000000000..04b1834529 --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystem.scala @@ -0,0 +1,51 @@ +/* + * 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. + */ + +package org.apache.texera.amber.util + +import org.apache.hadoop.fs.permission.FsPermission +import org.apache.hadoop.fs.{LocalFileSystem, Path, RawLocalFileSystem} +import org.apache.hadoop.util.Shell + +/** + * A Hadoop local file system that works without a `winutils.exe` installation. + * + * On Windows, Hadoop's default local file system implements chmod/chown by invoking + * `%HADOOP_HOME%\bin\winutils.exe`, and every file or directory creation applies POSIX + * permissions through that path. Without a native Hadoop installation this fails with + * "Hadoop bin directory does not exist". POSIX permission bits carry no meaning on NTFS, + * so on Windows hosts without winutils this file system skips permission operations + * instead of failing. + * + * Selected via `fs.file.impl` by [[IcebergUtil]] on Windows hosts where winutils is + * unavailable. Everywhere else — winutils installed, or a non-Windows host — it behaves + * exactly like Hadoop's default local file system, including its winutils/chmod use. + */ +class WinutilsFreeLocalFileSystem extends LocalFileSystem(new WinutilsFreeRawLocalFileSystem) + +class WinutilsFreeRawLocalFileSystem extends RawLocalFileSystem { + + private def skipPermissionOps: Boolean = Shell.WINDOWS && !Shell.hasWinutilsPath() + + override def setPermission(p: Path, permission: FsPermission): Unit = + if (!skipPermissionOps) super.setPermission(p, permission) + + override def setOwner(p: Path, username: String, groupname: String): Unit = + if (!skipPermissionOps) super.setOwner(p, username, groupname) +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/IcebergUtilSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/IcebergUtilSpec.scala index 75a997cbb5..44a75305d5 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/IcebergUtilSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/IcebergUtilSpec.scala @@ -21,13 +21,16 @@ package org.apache.texera.amber.util import org.apache.texera.amber.core.tuple.{AttributeType, LargeBinary, Schema, Tuple} import org.apache.texera.amber.util.IcebergUtil.toIcebergSchema +import org.apache.hadoop.fs.FileSystem import org.apache.iceberg.data.GenericRecord import org.apache.iceberg.exceptions.RESTException import org.apache.iceberg.types.Types import org.apache.iceberg.{Schema => IcebergSchema} import org.scalatest.flatspec.AnyFlatSpec +import java.net.URI import java.nio.ByteBuffer +import java.nio.file.Files import java.sql.Timestamp import java.time.{LocalDateTime, ZoneId} import scala.jdk.CollectionConverters._ @@ -300,6 +303,43 @@ class IcebergUtilSpec extends AnyFlatSpec { assert(IcebergUtil.fromRecord(record, schema) == tuple) } + it should "select WinutilsFreeLocalFileSystem when winutils is unavailable on Windows" in { + val conf = IcebergUtil.newLocalHadoopConf(useWinutilsFreeLocalFs = true) + assert(conf.get("fs.file.impl") == classOf[WinutilsFreeLocalFileSystem].getName) + assert(conf.getBoolean("fs.file.impl.disable.cache", false)) + assert( + FileSystem.get(URI.create("file:///"), conf).isInstanceOf[WinutilsFreeLocalFileSystem] + ) + } + + it should "leave the Hadoop configuration untouched when winutils is not needed" in { + val conf = IcebergUtil.newLocalHadoopConf(useWinutilsFreeLocalFs = false) + assert(conf.get("fs.file.impl") == null) + assert(!conf.getBoolean("fs.file.impl.disable.cache", false)) + } + + it should "create and load tables via createHadoopCatalog in a local warehouse on every platform" in { + val warehouse = Files.createTempDirectory("iceberg-local-warehouse") + val catalog = IcebergUtil.createHadoopCatalog("local_test", warehouse) + + // On Windows hosts without a winutils.exe installation, this used to fail with + // "Hadoop bin directory does not exist": Hadoop's default local file system + // shells out to winutils for chmod on every file/directory creation. + IcebergUtil.createTable( + catalog, + "test_namespace", + "test_table", + IcebergUtil.toIcebergSchema(Schema().add("id", AttributeType.INTEGER)), + overrideIfExists = true + ) + + assert( + Files.exists(warehouse.resolve("test_namespace").resolve("test_table").resolve("metadata")), + "table metadata must be written through the local file system" + ) + assert(IcebergUtil.loadTableMetadata(catalog, "test_namespace", "test_table").nonEmpty) + } + it should "surface RESTException when createRestCatalog cannot reach the REST endpoint" in { // Property Map is built before any network call. With or without // Lakekeeper reachable, .initialize surfaces a RESTException — the diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystemSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystemSpec.scala new file mode 100644 index 0000000000..217479b56f --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystemSpec.scala @@ -0,0 +1,84 @@ +/* + * 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. + */ + +package org.apache.texera.amber.util + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.permission.FsPermission +import org.apache.hadoop.fs.{FileSystem, Path} +import org.scalatest.flatspec.AnyFlatSpec + +import java.net.URI +import java.nio.charset.StandardCharsets +import java.nio.file.Files + +class WinutilsFreeLocalFileSystemSpec extends AnyFlatSpec { + + private def newFs(): FileSystem = { + val conf = new Configuration() + // Select the file system the same way production code does: through `fs.file.impl`. + conf.set("fs.file.impl", classOf[WinutilsFreeLocalFileSystem].getName) + conf.setBoolean("fs.file.impl.disable.cache", true) + FileSystem.get(URI.create("file:///"), conf) + } + + "WinutilsFreeLocalFileSystem" should "be selected through fs.file.impl" in { + assert(newFs().isInstanceOf[WinutilsFreeLocalFileSystem]) + } + + it should "create directories and read/write files without winutils" in { + val fs = newFs() + val tmp = Files.createTempDirectory("winutils-free-fs") + val dir = new Path(tmp.toUri.toString, "a/b/c") + assert(fs.mkdirs(dir)) + + val file = new Path(dir, "data.txt") + val out = fs.create(file, true) + out.write("hello".getBytes(StandardCharsets.UTF_8)) + out.close() + + val in = fs.open(file) + val buf = new Array[Byte](5) + in.readFully(buf) + in.close() + assert(new String(buf, StandardCharsets.UTF_8) == "hello") + + assert(fs.getFileStatus(file).getLen == 5) + assert(fs.delete(file, false)) + } + + it should "not fail on setPermission" in { + val fs = newFs() + val tmp = Files.createTempDirectory("winutils-free-fs-perm") + val file = new Path(tmp.toUri.toString, "perm.txt") + fs.create(file, true).close() + // No-op on Windows hosts without winutils; delegates to Hadoop's default elsewhere. + fs.setPermission(file, new FsPermission("755")) + } + + it should "not fail on setOwner" in { + val fs = newFs() + val tmp = Files.createTempDirectory("winutils-free-fs-owner") + val file = new Path(tmp.toUri.toString, "owner.txt") + fs.create(file, true).close() + // Chown-to-self is permitted everywhere; no-op on Windows hosts without winutils. + // (Not FileStatus.getOwner: reading the owner itself requires winutils on Windows.) + fs.setOwner(file, System.getProperty("user.name"), null) + } +}
