This is an automated email from the ASF dual-hosted git repository.
aglinxinyuan pushed a commit to branch release/v1.2
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/release/v1.2 by this push:
new 7bb46a7085 fix(workflow-core, v1.2): run Iceberg local storage on
Windows without winutils (#6528)
7bb46a7085 is described below
commit 7bb46a70854f128c99b78e71fb5ead252ba26181
Author: Xinyuan Lin <[email protected]>
AuthorDate: Tue Jul 21 17:20:19 2026 -0700
fix(workflow-core, v1.2): run Iceberg local storage on Windows without
winutils (#6528)
### What changes were proposed in this PR?
Backport of apache/texera#6488 to `release/v1.2`.
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.
`release/v1.2` carries the same regression: it is on hadoop-common
**3.4.3**, and 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 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. |
Clean cherry-pick of the squash commit (`aefae7b`) — no conflicts. The
change is confined to `common/workflow-core` Scala and adds no new
dependencies, so no `LICENSE-binary`/`NOTICE-binary` regeneration is
needed.
### Any related issues, documentation, discussions?
Backport of #6488 (Closes #6487).
### How was this PR tested?
- Both specs run green on `release/v1.2` with Java 17: `IcebergUtilSpec`
(local-warehouse create/load regression test + both branches of the
injectable winutils gate) and `WinutilsFreeLocalFileSystemSpec`
(selection via `fs.file.impl`, file round-trip,
`setPermission`/`setOwner` no-throw).
- `WorkflowCore/scalafmtCheckAll` and `WorkflowCore/scalafixAll --check`
pass.
No `release/*` label is added — this PR *is* the backport, so the
automated backport check correctly skips.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
---
.../org/apache/texera/amber/util/IcebergUtil.scala | 36 ++++++++-
.../amber/util/WinutilsFreeLocalFileSystem.scala | 51 +++++++++++++
.../apache/texera/amber/util/IcebergUtilSpec.scala | 48 ++++++++++++
.../util/WinutilsFreeLocalFileSystemSpec.scala | 89 ++++++++++++++++++++++
4 files changed, 221 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 0b45b9eec3..53b5079d5f 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.amber.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..a10f459073 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,51 @@ 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))
+ // Caching is disabled, so this is a fresh instance that must be closed to
avoid leaking handles.
+ val fs = FileSystem.get(URI.create("file:///"), conf)
+ try assert(fs.isInstanceOf[WinutilsFreeLocalFileSystem])
+ finally fs.close()
+ }
+
+ 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)
+ try {
+ // 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)
+ } finally {
+ // Close the catalog (and its underlying Hadoop FileSystem) to avoid
leaking handles across the suite.
+ catalog match {
+ case closeable: AutoCloseable => closeable.close()
+ case _ =>
+ }
+ }
+ }
+
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..e8d21f826b
--- /dev/null
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/WinutilsFreeLocalFileSystemSpec.scala
@@ -0,0 +1,89 @@
+/*
+ * 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)
+ }
+
+ // Caching is disabled, so every newFs() is a fresh instance that must be
closed to
+ // avoid leaking file handles across tests (especially on Windows).
Loan-pattern helper.
+ private def withFs(test: FileSystem => Unit): Unit = {
+ val fs = newFs()
+ try test(fs)
+ finally fs.close()
+ }
+
+ "WinutilsFreeLocalFileSystem" should "be selected through fs.file.impl" in
withFs { fs =>
+ assert(fs.isInstanceOf[WinutilsFreeLocalFileSystem])
+ }
+
+ it should "create directories and read/write files without winutils" in
withFs { fs =>
+ 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 withFs { fs =>
+ 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 withFs { fs =>
+ 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)
+ }
+}
