mattcasters commented on code in PR #8819:
URL: https://github.com/apache/hop/pull/8819#discussion_r4233145137
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java:
##########
@@ -40,7 +68,233 @@ public LakeTableOutput(
@Override
public boolean processRow() throws HopException {
- setOutputDone();
+ try (PluginClassLoader ignored = PluginClassLoader.activate()) {
+ return writeRow();
+ }
+ }
+
+ private boolean writeRow() throws HopException {
+ Object[] row = getRow();
+ if (first) {
+ first = false;
+ IRowMeta rowMeta =
+ row == null
+ ? getPipelineMeta().getPrevTransformFields(this,
getTransformMeta())
+ : getInputRowMeta();
+ data.coordinator = coordinator(rowMeta).orElse(null);
+ if (data.coordinator != null) {
+ data.writer =
+ new IcebergRowWriter(
+ data.coordinator.table(),
+ rowMeta,
+ getCopy(),
+ System.currentTimeMillis(),
+ IcebergRowWriter.DEFAULT_TARGET_FILE_SIZE);
+ }
+ }
+
+ if (row == null) {
+ if (data.writer != null) {
+ data.coordinator.addFiles(data.writer.complete());
+ data.writer = null;
+ }
+ setOutputDone();
+ return false;
+ }
+
+ if (data.writer != null) {
+ data.writer.write(row);
+ }
+ putRow(getInputRowMeta(), row);
+ return true;
+ }
+
+ /**
+ * The commit coordinator shared by all copies of this transform in this
run, created by the first
+ * copy that gets here. Empty when the save mode says to leave an existing
table alone.
+ */
+ @SuppressWarnings("unchecked")
+ private Optional<CommitCoordinator> coordinator(IRowMeta rowMeta) throws
HopException {
+ Map<String, Object> shared = getPipeline().getExtensionDataMap();
+ String key = COORDINATOR_KEY + getTransformName();
+ synchronized (shared) {
+ if (!shared.containsKey(key)) {
+ Optional<CommitCoordinator> coordinator =
Optional.ofNullable(newCoordinator(rowMeta));
+ shared.put(key, coordinator);
+ coordinator.ifPresent(this::commitWhenPipelineFinishes);
+ }
+ return (Optional<CommitCoordinator>) shared.get(key);
+ }
+ }
+
+ private CommitCoordinator newCoordinator(IRowMeta rowMeta) throws
HopException {
+ String format = resolve(meta.getFormat());
+ if (!LakeFormats.FORMAT_ICEBERG.equalsIgnoreCase(format)) {
+ throw new HopException(
+ "Table format '"
+ + format
+ + "' isn't supported on this engine yet. The local engine writes
Apache Iceberg"
+ + " tables; Delta Lake tables can be written with the Spark
engine.");
+ }
+
+ IcebergTableTarget target = target();
+ String saveMode =
+ StringUtils.defaultIfBlank(resolve(meta.getSaveMode()),
LakeTableOutputMeta.MODE_ERROR);
+ Map<String, String> summary = new HashMap<>();
+ summary.put(CommitCoordinator.SUMMARY_RUN_ID,
getPipeline().getLogChannelId());
+ summary.put("hop.pipeline", Const.NVL(getPipelineMeta().getName(), ""));
+
+ if (target.exists()) {
+ if (LakeTableOutputMeta.MODE_ERROR.equalsIgnoreCase(saveMode)) {
+ throw new HopException(
+ "Table "
+ + target.name()
+ + " already exists. Choose the Append or Overwrite save mode
to write to it.");
+ }
+ if (LakeTableOutputMeta.MODE_IGNORE.equalsIgnoreCase(saveMode)) {
+ logBasic("Table " + target.name() + " already exists, rows are not
written (Ignore)");
+ return null;
+ }
+ WriteMode mode;
+ if (LakeTableOutputMeta.MODE_OVERWRITE.equalsIgnoreCase(saveMode)) {
+ mode = WriteMode.OVERWRITE_TABLE;
+ } else if (LakeTableOutputMeta.MODE_APPEND.equalsIgnoreCase(saveMode)) {
+ mode = WriteMode.APPEND;
+ } else {
+ throw unsupportedSaveMode(saveMode);
+ }
+ return new CommitCoordinator(target.load(), mode, summary);
+ }
+
+ if (!isSupportedSaveMode(saveMode)) {
+ throw unsupportedSaveMode(saveMode);
+ }
+ Schema schema = IcebergTypeMapper.toIcebergSchema(rowMeta);
+ PartitionSpec spec = partitionSpec(schema,
resolve(meta.getPartitionByColumns()));
+ logBasic(
+ "Creating table "
+ + target.name()
+ + (spec.isUnpartitioned() ? "" : ", partitioned by " +
meta.getPartitionByColumns()));
+ return new CommitCoordinator(target.create(schema, spec, Map.of()),
summary);
+ }
+
+ private static boolean isSupportedSaveMode(String saveMode) {
+ for (String mode :
+ new String[] {
+ LakeTableOutputMeta.MODE_APPEND,
+ LakeTableOutputMeta.MODE_OVERWRITE,
+ LakeTableOutputMeta.MODE_ERROR,
+ LakeTableOutputMeta.MODE_IGNORE
+ }) {
+ if (mode.equalsIgnoreCase(saveMode)) {
+ return true;
+ }
+ }
return false;
}
+
+ private static HopException unsupportedSaveMode(String saveMode) {
+ return new HopException(
+ "Save mode '"
+ + saveMode
+ + "' isn't supported. Use Append, Overwrite, ErrorIfExists or
Ignore.");
+ }
+
+ private IcebergTableTarget target() throws HopException {
+ if
(LakeTableInputMeta.MODE_TABLE.equalsIgnoreCase(meta.getIdentifierMode())) {
+ String catalogName = resolve(meta.getCatalogMetadataName());
+ if (StringUtils.isBlank(catalogName)) {
+ throw new HopException("Table mode needs a catalog: please select
one");
+ }
+ LakeCatalog catalog =
metadataProvider.getSerializer(LakeCatalog.class).load(catalogName);
+ if (catalog == null) {
+ throw new HopException("Catalog '" + catalogName + "' couldn't be
found");
+ }
+ return IcebergTables.targetInCatalog(catalog, meta.getTableIdentifier(),
this);
+ }
+ String path = resolve(meta.getTablePath());
+ if (StringUtils.isBlank(path)) {
+ throw new HopException("Please specify the location of the table");
+ }
+ return IcebergTableTarget.atPath(path);
+ }
+
+ /** Identity partitions on the listed columns, used when the table is
created. */
+ static PartitionSpec partitionSpec(Schema schema, String columns) throws
HopException {
+ if (StringUtils.isBlank(columns)) {
+ return PartitionSpec.unpartitioned();
+ }
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(schema);
+ List<String> missing = new ArrayList<>();
+ for (String column : columns.split(",")) {
+ String name = column.trim();
+ if (name.isEmpty()) {
+ continue;
+ }
+ if (schema.findField(name) == null) {
+ missing.add(name);
+ } else {
+ builder.identity(name);
+ }
+ }
+ if (!missing.isEmpty()) {
+ throw new HopException("Partition column(s) not found in the input: " +
missing);
+ }
+ return builder.build();
+ }
+
+ private void commitWhenPipelineFinishes(CommitCoordinator coordinator) {
+ getPipeline()
+ .addExecutionFinishedListener(
Review Comment:
**[bug]** The snapshot is committed from an execution-finished listener,
which runs after Hop has already marked the pipeline finished. The built-in
listener is inserted at the front of the list and calls `setFinished(true)` and
`setRunning(false)` before this listener runs. `Pipeline.getStatus()` then
returns `Finished` while `finishRun` is still inside `coordinator.commit()`.
The remote engine, which this PR enables, polls that status and in
`RemotePipelineEngine` snapshots `getResult()` the moment it sees `Finished`,
then `waitUntilFinished()` returns. A parent workflow can therefore start the
next action before the snapshot exists, and a commit that fails later is not in
the result it already stored. On the local engine the window is smaller but
real: `fireExecutionFinishedListeners` calls `pipelineCompleted()` (which
unblocks `waitUntilFinished`) before the listener exception is applied with
`setErrors`. `ActionPipeline` checks `pipeline.getErrors()` immediately after
that wa
it, so a failed commit that deleted the run's files can still be reported as
success.
**Suggestion:** Record the failure on the transform with `setErrors` inside
`finishRun` before returning, so local waiters observe it before
`pipelineCompleted()`. For the status flag, keep the pipeline reporting
`Running` for the duration of the commit (`setRunning(true)` on entry and
`setRunning(false)` plus the error count on exit) so a remote status poll
cannot observe `Finished` with zero errors until the snapshot attempt has
ended. Add a test that a commit failure leaves `getErrors() > 0` after
`waitUntilFinished()` returns.
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.hop.lakehouse.iceberg;
+
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.FileSystemException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.vfs2.FileObject;
+import org.apache.commons.vfs2.provider.local.LocalFile;
+import org.apache.hop.core.logging.LogChannel;
+import org.apache.hop.core.vfs.HopVfs;
+import org.apache.hop.lakehouse.iceberg.io.HopVfsFileIO;
+import org.apache.iceberg.LocationProviders;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+
+/**
+ * Table operations for an Iceberg table that lives in a folder without a
catalog, in the layout of
+ * Hadoop catalogs and Spark path tables: {@code metadata/v<N>.metadata.json}
plus a {@code
+ * version-hint.text} that names the current version.
+ *
+ * <p>A commit writes the new metadata to a temporary file and then publishes
it under the next
+ * version's name with an operation that fails if that name is already taken.
On a local file system
+ * that is a hard link, which the operating system creates atomically, so two
writers can't both
+ * commit the same version, whether they run in one Hop instance or in several
processes. Commits in
+ * one JVM are also serialized per table location. Other file systems publish
with a check followed
+ * by a move, which isn't atomic across processes: like Iceberg's Hadoop
tables on S3, a path table
+ * there should have one writer at a time. Tables managed by a catalog should
be written through the
+ * catalog instead.
+ *
+ * <p>The version hint is best effort. Once the metadata file is in place the
commit has happened,
+ * so a failure to update the hint is only logged, and readers walk forward
from the hint to the
+ * newest version (see {@link IcebergTables#findCurrentMetadataFile(String)}).
+ */
+public class PathTableOperations implements TableOperations {
+
+ /** Serializes commits to the same table location within this JVM. */
+ private static final Map<String, ReentrantLock> LOCKS = new
ConcurrentHashMap<>();
+
+ private final String location;
+ private final FileIO io = new HopVfsFileIO();
+ private TableMetadata current;
+ private int version = -1;
+
+ public PathTableOperations(String location) {
+ this.location =
+ location.endsWith("/") ? location.substring(0, location.length() - 1)
: location;
+ }
+
+ @Override
+ public TableMetadata current() {
+ if (current == null && version < 0) {
+ refresh();
+ }
+ return current;
+ }
+
+ @Override
+ public TableMetadata refresh() {
+ String metadataFile;
+ try {
+ metadataFile = IcebergTables.findCurrentMetadataFile(location);
+ } catch (Exception e) {
+ throw new RuntimeIOException(
+ new java.io.IOException("Unable to read Iceberg table at " +
location, e));
+ }
+ if (metadataFile == null) {
+ // No table yet: it is created by the first commit.
+ version = 0;
+ current = null;
+ } else {
+ version =
+ (int)
+ IcebergTables.metadataVersion(
+ metadataFile.substring(metadataFile.lastIndexOf('/') + 1));
+ current = TableMetadataParser.read(io, metadataFile);
+ }
+ return current;
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ ReentrantLock lock = LOCKS.computeIfAbsent(location, ignored -> new
ReentrantLock());
+ lock.lock();
+ try {
+ commitLocked(base, metadata);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void commitLocked(TableMetadata base, TableMetadata metadata) {
+ TableMetadata latest = refresh();
+ if (base != latest
+ && (base == null
+ || latest == null
+ ||
!base.metadataFileLocation().equals(latest.metadataFileLocation()))) {
+ throw new CommitFailedException(
+ "Table %s was changed by another writer since it was read",
location);
+ }
+
+ int next = version + 1;
+ String target = metadataFileLocation("v" + next + ".metadata.json");
+ String temp = metadataFileLocation(UUID.randomUUID() +
".metadata.json.tmp");
+ TableMetadataParser.write(metadata, io.newOutputFile(temp));
+ try {
+ publish(temp, target, next);
+ } finally {
+ deleteQuietly(temp);
+ }
+
+ // The commit has happened. Nothing after this point may fail it.
+ try {
+ writeVersionHint(next);
+ } catch (Exception e) {
+ LogChannel.GENERAL.logBasic(
+ "Committed version "
+ + next
+ + " of Iceberg table "
+ + location
+ + ", but couldn't update version-hint.text; readers find the new
version anyway: "
+ + e.getMessage());
+ }
+ // Read the new metadata lazily, on the next call to current() or
refresh().
+ current = null;
+ version = -1;
+ }
+
+ /**
+ * Publishes {@code temp} as {@code target}, failing with a {@link
CommitFailedException} if
+ * another writer committed that version first.
+ */
+ void publish(String temp, String target, int next) {
+ try {
+ FileObject tempFile = HopVfs.getFileObject(temp);
+ FileObject targetFile = HopVfs.getFileObject(target);
+ if (tempFile instanceof LocalFile && targetFile instanceof LocalFile) {
+ Path tempPath = Paths.get(tempFile.getURI());
+ Path targetPath = Paths.get(targetFile.getURI());
+ try {
+ // link(2) fails with EEXIST if the name is taken, atomically.
+ Files.createLink(targetPath, tempPath);
+ } catch (UnsupportedOperationException | FileSystemException
linkNotSupported) {
+ if (linkNotSupported instanceof FileAlreadyExistsException) {
+ throw linkNotSupported;
+ }
+ // No hard links on this file system: a move that refuses an
existing target.
+ Files.move(tempPath, targetPath);
+ }
+ return;
+ }
+ if (targetFile.exists()) {
+ throw new FileAlreadyExistsException(target);
+ }
+ tempFile.moveTo(targetFile);
Review Comment:
**[bug]** On every file system that is not a `LocalFile`, `publish` still
claims it fails when `v<N>.metadata.json` is already taken, but
`FileObject.moveTo` does the opposite. Commons VFS `AbstractFileObject.moveTo`
calls `destFile.deleteAll()` when the destination exists, and the `exists()`
check above is not atomic with that delete. Two writers (two Hop servers, or
Hop and Spark on the same path table) can both pass the check; the later
`moveTo` then deletes the earlier version file and both commits report success.
That drops a snapshot that was already published. The in-JVM lock does not
cover this. The local fallback at line 176 has the same shape: `Files.move`
without options ends in `rename(2)`, which replaces an existing name, so a lost
race on a filesystem where `createLink` is unavailable also overwrites the
committed version instead of failing.
**Suggestion:** For non-local providers, publish with an operation that
fails if the destination exists (HDFS `rename` that does not delete the target,
S3 conditional put / If-None-Match) and treat a failure after the file may
already be visible as `CommitStateUnknownException`. If that is out of scope,
do not call `moveTo` at all: fail the commit when the target exists and
document that path tables are single-writer on every non-local filesystem,
including HDFS. On the local fallback, use `Files.move` with `ATOMIC_MOVE` only
where the filesystem rejects an existing target, or keep the hard link as the
only success path.
##########
plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java:
##########
@@ -40,7 +68,233 @@ public LakeTableOutput(
@Override
public boolean processRow() throws HopException {
- setOutputDone();
+ try (PluginClassLoader ignored = PluginClassLoader.activate()) {
+ return writeRow();
+ }
+ }
+
+ private boolean writeRow() throws HopException {
+ Object[] row = getRow();
+ if (first) {
+ first = false;
+ IRowMeta rowMeta =
+ row == null
+ ? getPipelineMeta().getPrevTransformFields(this,
getTransformMeta())
+ : getInputRowMeta();
+ data.coordinator = coordinator(rowMeta).orElse(null);
+ if (data.coordinator != null) {
+ data.writer =
+ new IcebergRowWriter(
+ data.coordinator.table(),
+ rowMeta,
+ getCopy(),
+ System.currentTimeMillis(),
+ IcebergRowWriter.DEFAULT_TARGET_FILE_SIZE);
+ }
+ }
+
+ if (row == null) {
+ if (data.writer != null) {
+ data.coordinator.addFiles(data.writer.complete());
+ data.writer = null;
+ }
+ setOutputDone();
+ return false;
+ }
+
+ if (data.writer != null) {
+ data.writer.write(row);
+ }
+ putRow(getInputRowMeta(), row);
+ return true;
+ }
+
+ /**
+ * The commit coordinator shared by all copies of this transform in this
run, created by the first
+ * copy that gets here. Empty when the save mode says to leave an existing
table alone.
+ */
+ @SuppressWarnings("unchecked")
+ private Optional<CommitCoordinator> coordinator(IRowMeta rowMeta) throws
HopException {
+ Map<String, Object> shared = getPipeline().getExtensionDataMap();
+ String key = COORDINATOR_KEY + getTransformName();
+ synchronized (shared) {
+ if (!shared.containsKey(key)) {
+ Optional<CommitCoordinator> coordinator =
Optional.ofNullable(newCoordinator(rowMeta));
+ shared.put(key, coordinator);
+ coordinator.ifPresent(this::commitWhenPipelineFinishes);
+ }
+ return (Optional<CommitCoordinator>) shared.get(key);
+ }
+ }
+
+ private CommitCoordinator newCoordinator(IRowMeta rowMeta) throws
HopException {
+ String format = resolve(meta.getFormat());
Review Comment:
**[suggestion]** Coalesce and the extra Spark options are still on the
dialog and change a Spark write, but this path never reads
`meta.getCoalescePartitions()` or `meta.getExtraOptions()`. Moving the same
pipeline to the local or remote engine silently ignores them (target file size,
a dynamic overwrite option, and anything else stored there) and still reports
success. `Map.of()` below also drops table properties the user may have set for
the Spark engine.
**Suggestion:** If either value is non-blank, log a warning or fail the
transform with the option name, and only pass through properties this writer
actually implements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
