You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by mc...@apache.org on 2026/10/10 08:19:12 UTC
(hop) branch main updated: Issue #5524 : Write Iceberg tables with Lake table output on the local engine (#8819)
This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 45aaf23ca0 Issue #5524 : Write Iceberg tables with Lake table output on the local engine (#8819)
45aaf23ca0 is described below
commit 45aaf23ca03732e433fdbcdcb2c204e2f7f09e27
Author: Ankur Partap Kotwal <40...@users.noreply.github.com>
AuthorDate: Sat Oct 10 01:19:06 2026 -0700
Issue #5524 : Write Iceberg tables with Lake table output on the local engine (#8819)
* Issue #5524 : Write Iceberg tables with Lake table output on the local engine
Lake table output now runs on the local and remote engines for Apache
Iceberg tables, with the same save modes as on the Spark engine:
Append and Overwrite write to an existing table, Ignore leaves it alone,
ErrorIfExists refuses it, and every mode creates a missing table from
the incoming row layout and the partition columns.
Each copy of the transform writes its own Parquet files through Hop VFS.
The files of a run are committed as one snapshot when the whole pipeline
has finished without errors, and deleted otherwise, so a transform that
fails further down the pipeline still rolls the run back. A new table is
created in the same commit as its first snapshot, so a failed run leaves
no empty table behind. Overwrites check that no other writer changed the
table since the run started.
Path tables are written in the layout of Hadoop catalogs and Spark path
tables (v<N>.metadata.json and version-hint.text), so both engines can
write the same table. Tables at a path that belong to a catalog are
refused: they need to be written through their catalog.
Generated-by: Claude Opus 5.5
* Issue #5524 : Address review of the local Iceberg write
- Path-table commits can no longer replace each other. On a local file
system the new version is published with a hard link, which fails if
the name is taken; without hard links, with a move that refuses an
existing target. Commits in one JVM are also serialized per table
location. Other file systems keep the check-and-move, so path tables
there should have one writer at a time, as documented.
- version-hint.text is a starting point, not the answer: the lookup
walks forward through v<N+1>, v<N+2>, ... (also .gz.metadata.json),
like Iceberg's HadoopTableOperations, and an unreadable hint falls
back to the newest file. A stale hint no longer hides a newer version
or blocks every later commit.
- Once the metadata file is in place the commit has happened: a failure
to update the hint is logged, the table is read back lazily, and the
run's data files are never deleted. CommitStateUnknownException keeps
the files too and fails the run with a message to check the table.
- Save modes other than Append, Overwrite, ErrorIfExists and Ignore are
rejected instead of appending.
Generated-by: Claude Opus 5.5
* Issue #5524 : Address second review of the local Iceberg write
- Publishing a path-table version never deletes or replaces an existing
file anymore. A hard link is the only atomic route, on local file
systems that support it. Everywhere else (HDFS, object stores, local
file systems without hard links) the version file is written directly
after an exists check instead of with VFS moveTo, which deleted an
existing destination; a failed write there is reported as
CommitStateUnknownException. Path tables on those file systems are
documented as single-writer.
- The commit runs from an execution-finished listener, after Hop has
marked the pipeline finished. The pipeline now reports itself as
running for the duration of the commit, so a status poll (remote
engine, parent workflow) can't see Finished before the snapshot
exists, and a failed commit is counted as a transform error before the
listener returns, so it is in the result waiters read.
- Coalesce partitions and Extra options, which only configure a Spark
write, are reported as ignored when set on the local engine.
Generated-by: Claude Opus 5.5
* Issue #5524 : Address third review of the local Iceberg write
- The commit listener is inserted in front of Hop's own
execution-finished listener, so the commit runs while the pipeline
still reports Running: a status poll can no longer see Finished before
the snapshot exists. This replaces the setRunning workaround, which ran
too late. A failed commit is still counted as a transform error before
the listener returns.
- Overwrites always validate against concurrent data, also when the
table had no snapshot when the run started, so a writer that gives an
empty table its first snapshot during the run makes the overwrite fail
instead of being removed by it.
- Local publish is atomic whether or not hard links are available: a
hard link, else an exclusive create (CREATE_NEW). Only a missing hard
link support falls through; other errors fail the commit. A partial
version file from a failed write is removed, locally and on other file
systems, so readers never pick it up.
- The legacy gzip name v<N>.metadata.json.gz is recognized, both when
walking forward from the version hint and in the folder listing.
Generated-by: Claude Opus 5.5
---
.../hop/lakehouse/iceberg/CommitCoordinator.java | 198 ++++++++
.../hop/lakehouse/iceberg/IcebergRowWriter.java | 156 ++++++
.../hop/lakehouse/iceberg/IcebergTableTarget.java | 123 +++++
.../hop/lakehouse/iceberg/IcebergTables.java | 85 +++-
.../hop/lakehouse/iceberg/PathTableOperations.java | 294 +++++++++++
.../hop/lakehouse/transforms/LakeTableOutput.java | 299 ++++++++++-
.../lakehouse/transforms/LakeTableOutputData.java | 9 +
.../lakehouse/transforms/LakeTableOutputMeta.java | 6 +-
.../lakehouse/iceberg/CommitCoordinatorTest.java | 95 ++++
.../hop/lakehouse/iceberg/IcebergTablesTest.java | 14 +-
.../lakehouse/iceberg/PathTableOperationsTest.java | 253 ++++++++++
.../transforms/LakeTableOutputLocalTest.java | 554 +++++++++++++++++++++
12 files changed, 2063 insertions(+), 23 deletions(-)
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/CommitCoordinator.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/CommitCoordinator.java
new file mode 100644
index 0000000000..dc87087d07
--- /dev/null
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/CommitCoordinator.java
@@ -0,0 +1,198 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.ReplacePartitions;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotUpdate;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Collects the data files written by every copy of an output transform and turns them into a single
+ * snapshot, or into nothing at all.
+ *
+ * <p>Copies call {@link #addFiles(List)} as they finish. Once the whole pipeline is done, exactly
+ * one of {@link #commit()} or {@link #abort()} is called: commit when the run ended without errors,
+ * abort otherwise. This keeps a run atomic. Readers either see all of its rows or none of them,
+ * even when a transform further down the pipeline fails after this one has finished writing.
+ */
+public class CommitCoordinator {
+
+ public enum WriteMode {
+ APPEND,
+ OVERWRITE_TABLE,
+ OVERWRITE_PARTITIONS
+ }
+
+ public static final String SUMMARY_RUN_ID = "hop.run-id";
+
+ private final Table table;
+ private final Transaction createTransaction;
+ private final WriteMode writeMode;
+ private final Map<String, String> summary;
+ private final Long startSnapshotId;
+ private boolean published;
+ private boolean outcomeUnknown;
+ private final List<DataFile> files = new ArrayList<>();
+ private boolean done;
+
+ /**
+ * Commits to an existing table.
+ *
+ * @param summary extra properties for the snapshot summary, such as the pipeline name and run id
+ */
+ public CommitCoordinator(Table table, WriteMode writeMode, Map<String, String> summary) {
+ this(table, null, writeMode, summary);
+ }
+
+ /**
+ * Commits to a table that is created by this run. The table only comes into existence together
+ * with the run's snapshot, so a failed run leaves no empty table behind.
+ */
+ public CommitCoordinator(Transaction createTransaction, Map<String, String> summary) {
+ this(createTransaction.table(), createTransaction, WriteMode.APPEND, summary);
+ }
+
+ private CommitCoordinator(
+ Table table,
+ Transaction createTransaction,
+ WriteMode writeMode,
+ Map<String, String> summary) {
+ this.table = table;
+ this.createTransaction = createTransaction;
+ this.writeMode = writeMode;
+ this.summary = Map.copyOf(summary);
+ Snapshot current = createTransaction == null ? table.currentSnapshot() : null;
+ this.startSnapshotId = current == null ? null : current.snapshotId();
+ }
+
+ /** The table to write data files for. */
+ public Table table() {
+ return table;
+ }
+
+ public synchronized void addFiles(List<DataFile> dataFiles) {
+ if (done) {
+ throw new IllegalStateException("The commit for this run has already been decided");
+ }
+ files.addAll(dataFiles);
+ }
+
+ /**
+ * Commits all collected files as one snapshot.
+ *
+ * @return the id of the new snapshot, or null if the commit happened but the table couldn't be
+ * read back afterwards
+ * @throws CommitStateUnknownException if the catalog couldn't say whether the commit happened;
+ * the data files must then be kept, see {@link #isPublished()}
+ */
+ public synchronized Long commit() {
+ done = true;
+ SnapshotUpdate<?> update =
+ switch (writeMode) {
+ case APPEND -> {
+ AppendFiles append = table.newAppend();
+ files.forEach(append::appendFile);
+ yield append;
+ }
+ case OVERWRITE_TABLE -> {
+ OverwriteFiles overwrite =
+ table.newOverwrite().overwriteByRowFilter(Expressions.alwaysTrue());
+ files.forEach(overwrite::addFile);
+ // Fail instead of silently replacing data another writer added in the meantime. With
+ // no starting snapshot (an empty table), every file now in the table counts as added
+ // since the start, so a writer that gave the table its first snapshot is caught too.
+ if (startSnapshotId != null) {
+ overwrite.validateFromSnapshot(startSnapshotId);
+ }
+ overwrite.validateNoConflictingData();
+ yield overwrite;
+ }
+ case OVERWRITE_PARTITIONS -> {
+ ReplacePartitions replace = table.newReplacePartitions();
+ files.forEach(replace::addFile);
+ if (startSnapshotId != null) {
+ replace.validateFromSnapshot(startSnapshotId);
+ }
+ replace.validateNoConflictingData().validateNoConflictingDeletes();
+ yield replace;
+ }
+ };
+ summary.forEach(update::set);
+ try {
+ update.commit();
+ if (createTransaction != null) {
+ createTransaction.commitTransaction();
+ }
+ } catch (CommitStateUnknownException e) {
+ outcomeUnknown = true;
+ throw e;
+ }
+ published = true;
+
+ // The snapshot is in the table from here on: reading it back is only for the log.
+ try {
+ Table committed = createTransaction != null ? createTransaction.table() : table;
+ committed.refresh();
+ Snapshot snapshot = committed.currentSnapshot();
+ return snapshot == null ? null : snapshot.snapshotId();
+ } catch (RuntimeException e) {
+ return null;
+ }
+ }
+
+ /** True once the run's snapshot is in the table; the data files must then never be deleted. */
+ public synchronized boolean isPublished() {
+ return published;
+ }
+
+ /** True if the catalog couldn't say whether the commit happened. */
+ public synchronized boolean isOutcomeUnknown() {
+ return outcomeUnknown;
+ }
+
+ /**
+ * Deletes every collected data file, so nothing of this run becomes visible. Refuses to do so
+ * once the snapshot is published or when the commit outcome is unknown, since the table may then
+ * reference those files.
+ */
+ public synchronized void abort() {
+ if (published || outcomeUnknown) {
+ throw new IllegalStateException(
+ "The data files of this run may be referenced by the table and are kept");
+ }
+ done = true;
+ for (DataFile file : files) {
+ table.io().deleteFile(file.location());
+ }
+ files.clear();
+ }
+
+ public synchronized int fileCount() {
+ return files.size();
+ }
+}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergRowWriter.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergRowWriter.java
new file mode 100644
index 0000000000..0e0040acfc
--- /dev/null
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergRowWriter.java
@@ -0,0 +1,156 @@
+/*
+ * 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.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericFileWriterFactory;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.InternalRecordWrapper;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileWriterFactory;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.PartitionedFanoutWriter;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.io.UnpartitionedWriter;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Writes Hop rows into new data files of an Iceberg table. It doesn't commit anything: {@link
+ * #complete()} returns the files, and a {@link CommitCoordinator} decides whether they become part
+ * of a snapshot. One writer is used per transform copy.
+ */
+public class IcebergRowWriter implements Closeable {
+
+ /** Iceberg's default target data file size. */
+ public static final long DEFAULT_TARGET_FILE_SIZE = 512L * 1024 * 1024;
+
+ private final Schema schema;
+ private final List<Types.NestedField> columns;
+ private final int[] rowIndexes;
+ private final IValueMeta[] valueMetas;
+ private final TaskWriter<Record> writer;
+ private boolean finished;
+
+ /**
+ * @param table the target table
+ * @param rowMeta layout of the incoming Hop rows
+ * @param copyNr transform copy number, used to keep file names unique across copies
+ * @param runId unique id of this pipeline run, also used in file names
+ * @param targetFileSize data files are rolled over at about this size
+ */
+ public IcebergRowWriter(
+ Table table, IRowMeta rowMeta, int copyNr, long runId, long targetFileSize)
+ throws HopException {
+ this.schema = table.schema();
+ this.columns = schema.columns();
+ this.rowIndexes = new int[columns.size()];
+ this.valueMetas = new IValueMeta[columns.size()];
+ for (int i = 0; i < columns.size(); i++) {
+ String name = columns.get(i).name();
+ int index = rowMeta.indexOfValue(name);
+ if (index < 0 && columns.get(i).isRequired()) {
+ throw new HopException(
+ "Required column '" + name + "' of table " + table.name() + " is missing in the input");
+ }
+ rowIndexes[i] = index;
+ valueMetas[i] = index < 0 ? null : rowMeta.getValueMeta(index);
+ }
+
+ FileIO io = table.io();
+ PartitionSpec spec = table.spec();
+ FileWriterFactory<Record> writerFactory =
+ new GenericFileWriterFactory.Builder(table).dataFileFormat(FileFormat.PARQUET).build();
+ OutputFileFactory fileFactory =
+ OutputFileFactory.builderFor(table, copyNr, runId).format(FileFormat.PARQUET).build();
+
+ if (spec.isUnpartitioned()) {
+ writer =
+ new UnpartitionedWriter<>(
+ spec, FileFormat.PARQUET, writerFactory, fileFactory, io, targetFileSize);
+ } else {
+ PartitionKey partitionKey = new PartitionKey(spec, schema);
+ InternalRecordWrapper wrapper = new InternalRecordWrapper(schema.asStruct());
+ writer =
+ new PartitionedFanoutWriter<>(
+ spec, FileFormat.PARQUET, writerFactory, fileFactory, io, targetFileSize) {
+ @Override
+ protected PartitionKey partition(Record row) {
+ partitionKey.partition(wrapper.wrap(row));
+ return partitionKey;
+ }
+ };
+ }
+ }
+
+ public void write(Object[] row) throws HopException {
+ GenericRecord record = GenericRecord.create(schema);
+ for (int i = 0; i < columns.size(); i++) {
+ if (rowIndexes[i] >= 0) {
+ Object value = row[rowIndexes[i]];
+ record.set(
+ i, IcebergTypeMapper.toIcebergValue(valueMetas[i], value, columns.get(i).type()));
+ }
+ }
+ try {
+ writer.write(record);
+ } catch (IOException e) {
+ throw new HopException("Error writing a row to the Iceberg table", e);
+ }
+ }
+
+ /** Closes the open files and returns every data file written. */
+ public List<DataFile> complete() throws HopException {
+ finished = true;
+ try {
+ return Arrays.asList(writer.dataFiles());
+ } catch (IOException e) {
+ throw new HopException("Error finishing the Iceberg data files", e);
+ }
+ }
+
+ /** Closes the writer and deletes the files it wrote. Used when the pipeline fails. */
+ public void abort() throws HopException {
+ finished = true;
+ try {
+ writer.abort();
+ } catch (IOException e) {
+ throw new HopException("Error cleaning up Iceberg data files", e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!finished) {
+ writer.abort();
+ finished = true;
+ }
+ }
+}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTableTarget.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTableTarget.java
new file mode 100644
index 0000000000..f3389fbee8
--- /dev/null
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTableTarget.java
@@ -0,0 +1,123 @@
+/*
+ * 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.util.Map;
+import org.apache.hop.core.exception.HopException;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.Transactions;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+/** A table to write to: one that may exist already, or that a write creates. */
+public interface IcebergTableTarget {
+
+ /** A readable name for messages. */
+ String name();
+
+ boolean exists() throws HopException;
+
+ /** The table, to write to. */
+ Table load() throws HopException;
+
+ /** The table, to read from. */
+ default Table read() throws HopException {
+ return load();
+ }
+
+ /**
+ * Starts creating the table. Nothing is visible until the transaction is committed, together with
+ * the first snapshot.
+ */
+ Transaction create(Schema schema, PartitionSpec spec, Map<String, String> properties)
+ throws HopException;
+
+ /** A table in a folder, in the layout of Hadoop catalogs and Spark path tables. */
+ static IcebergTableTarget atPath(String location) {
+ PathTableOperations ops = new PathTableOperations(location);
+ return new IcebergTableTarget() {
+ @Override
+ public String name() {
+ return ops.location();
+ }
+
+ @Override
+ public boolean exists() {
+ return ops.current() != null;
+ }
+
+ @Override
+ public Table load() throws HopException {
+ if (!ops.isPathTable()) {
+ throw new HopException(
+ "The Iceberg table at "
+ + ops.location()
+ + " is managed by a catalog. Write to it through that catalog (table mode), so"
+ + " the catalog sees the change.");
+ }
+ return new BaseTable(ops, ops.location());
+ }
+
+ @Override
+ public Table read() throws HopException {
+ return IcebergTables.loadFromPath(ops.location());
+ }
+
+ @Override
+ public Transaction create(Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(schema, spec, ops.location(), properties);
+ return Transactions.createTableTransaction(ops.location(), ops, metadata);
+ }
+ };
+ }
+
+ /** A table in a catalog. */
+ static IcebergTableTarget inCatalog(Catalog catalog, TableIdentifier identifier) {
+ return new IcebergTableTarget() {
+ @Override
+ public String name() {
+ return catalog.name() + "." + identifier;
+ }
+
+ @Override
+ public boolean exists() {
+ return catalog.tableExists(identifier);
+ }
+
+ @Override
+ public Table load() {
+ return catalog.loadTable(identifier);
+ }
+
+ @Override
+ public Transaction create(Schema schema, PartitionSpec spec, Map<String, String> properties) {
+ return catalog
+ .buildTable(identifier, schema)
+ .withPartitionSpec(spec)
+ .withProperties(properties)
+ .createTransaction();
+ }
+ };
+ }
+}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java
index 13f45fce0b..0211332f4f 100644
--- a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java
@@ -46,7 +46,7 @@ public final class IcebergTables {
/** Hadoop catalogs and Spark path tables name metadata files v1.metadata.json, v2... */
private static final Pattern HADOOP_METADATA =
- Pattern.compile("v(\\d+)(\\.gz)?\\.metadata\\.json");
+ Pattern.compile("v(\\d+)(?:(?:\\.gz)?\\.metadata\\.json|\\.metadata\\.json\\.gz)");
/** Other catalogs name them 00001-<uuid>.metadata.json, 00002-... */
private static final Pattern CATALOG_METADATA =
@@ -75,6 +75,19 @@ public final class IcebergTables {
*/
public static Table loadFromCatalog(
LakeCatalog catalog, String tableIdentifier, IVariables variables) throws HopException {
+ IcebergTableTarget target = targetInCatalog(catalog, tableIdentifier, variables);
+ if (!target.exists()) {
+ throw new HopException("Table " + target.name() + " doesn't exist");
+ }
+ return target.read();
+ }
+
+ /**
+ * The table {@code tableIdentifier} in the catalog described by a lakehouse catalog metadata
+ * object, which may or may not exist yet.
+ */
+ public static IcebergTableTarget targetInCatalog(
+ LakeCatalog catalog, String tableIdentifier, IVariables variables) throws HopException {
if (catalog == null) {
throw new HopException("No catalog specified to look up table '" + tableIdentifier + "'");
}
@@ -90,13 +103,13 @@ public final class IcebergTables {
throw new HopException(
"Catalog '" + catalog.getName() + "' (hadoop) needs a warehouse location");
}
- return loadFromPath(hadoopTableLocation(warehouse, identifier));
+ return IcebergTableTarget.atPath(hadoopTableLocation(warehouse, identifier));
}
case LakeCatalog.TYPE_REST -> {
Catalog rest =
CatalogUtil.loadCatalog(
REST_CATALOG, catalogName, restProperties(catalog, variables), null);
- return rest.loadTable(identifier);
+ return IcebergTableTarget.inCatalog(rest, identifier);
}
default ->
throw new HopException(
@@ -189,27 +202,52 @@ public final class IcebergTables {
return properties;
}
- /** The newest metadata file of the table at {@code root}. */
+ /** The current metadata file of the table at {@code location}. */
static String currentMetadataFile(String location) throws HopException {
+ String metadataFile = findCurrentMetadataFile(location);
+ if (metadataFile == null) {
+ throw new HopException(
+ "No Iceberg table found at '" + location + "': there is no table metadata file");
+ }
+ return metadataFile;
+ }
+
+ /**
+ * The current metadata file of the table at {@code location}, or null if there is no table there.
+ *
+ * <p>The version hint is only a starting point: a writer moves {@code v<N>.metadata.json} into
+ * place before it updates the hint, so after a crash between the two, or a failed hint write, the
+ * hint names an older version. Like Iceberg's {@code HadoopTableOperations}, the lookup starts at
+ * the hinted version and walks forward through {@code v<N+1>}, {@code v<N+2>}, ... to the newest
+ * one that exists. Without a usable hint, the newest metadata file in the folder is used.
+ */
+ static String findCurrentMetadataFile(String location) throws HopException {
String root = StringUtils.removeEnd(location, "/");
String metadataFolder = root + "/metadata";
try {
FileObject hint = HopVfs.getFileObject(metadataFolder + "/version-hint.text");
if (hint.exists()) {
+ long hinted = -1;
try (InputStream in = HopVfs.getInputStream(hint)) {
- String version = new String(in.readAllBytes(), StandardCharsets.UTF_8).trim();
- FileObject file =
- HopVfs.getFileObject(metadataFolder + "/v" + version + ".metadata.json");
- if (file.exists()) {
- return metadataFolder + "/v" + version + ".metadata.json";
+ hinted = Long.parseLong(new String(in.readAllBytes(), StandardCharsets.UTF_8).trim());
+ } catch (Exception e) {
+ // An unreadable hint is ignored, the folder listing below finds the newest version.
+ }
+ String file = hinted < 0 ? null : versionFile(metadataFolder, hinted);
+ if (file != null) {
+ for (long next = hinted + 1; ; next++) {
+ String newer = versionFile(metadataFolder, next);
+ if (newer == null) {
+ return file;
+ }
+ file = newer;
}
}
}
FileObject folder = HopVfs.getFileObject(metadataFolder);
if (!folder.exists()) {
- throw new HopException(
- "No Iceberg table found at '" + root + "': there is no metadata folder");
+ return null;
}
String newest = null;
long newestVersion = -1;
@@ -221,18 +259,29 @@ public final class IcebergTables {
newest = name;
}
}
- if (newest == null) {
- throw new HopException(
- "No Iceberg table found at '" + root + "': the metadata folder has no metadata files");
- }
- return metadataFolder + "/" + newest;
- } catch (HopException e) {
- throw e;
+ return newest == null ? null : metadataFolder + "/" + newest;
} catch (Exception e) {
throw new HopException("Unable to find the Iceberg metadata of table '" + root + "'", e);
}
}
+ /** The metadata file of {@code version} in a path-table layout, or null if it doesn't exist. */
+ private static String versionFile(String metadataFolder, long version) throws Exception {
+ // Iceberg's three spellings: plain, gzip, and the legacy gzip name (.metadata.json.gz).
+ for (String name :
+ new String[] {
+ "v" + version + ".metadata.json",
+ "v" + version + ".gz.metadata.json",
+ "v" + version + ".metadata.json.gz"
+ }) {
+ String file = metadataFolder + "/" + name;
+ if (HopVfs.getFileObject(file).exists()) {
+ return file;
+ }
+ }
+ return null;
+ }
+
/** Version number in a metadata file name, or -1 if it isn't a table metadata file. */
static long metadataVersion(String fileName) {
for (Pattern pattern : new Pattern[] {HADOOP_METADATA, CATALOG_METADATA}) {
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java
new file mode 100644
index 0000000000..b48376fc9a
--- /dev/null
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java
@@ -0,0 +1,294 @@
+/*
+ * 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.InputStream;
+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.CommitStateUnknownException;
+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. Where hard links aren't supported, a local commit
+ * uses an exclusive create instead, which is atomic too. Other file systems publish with a check
+ * followed by a write, which isn't atomic across processes, so a path table there (including HDFS,
+ * object stores and local file systems without hard links) should have one writer at a time. Tables
+ * managed by a catalog should be written through the catalog instead. Nothing in a commit ever
+ * deletes or replaces an existing version file.
+ *
+ * <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. Nothing here ever deletes or replaces an existing
+ * version file, and a failed write never leaves a partial one behind.
+ *
+ * <p>On a local disk this is atomic, also between processes: a hard link, or where hard links
+ * aren't supported, an exclusive create ({@code O_EXCL}) of the target. Everywhere else (HDFS,
+ * object stores) Hop VFS has no exclusive create, so the target is checked and then written: path
+ * tables there support one writer at a time.
+ */
+ 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) {
+ publishLocal(Paths.get(tempFile.getURI()), Paths.get(targetFile.getURI()));
+ } else {
+ publishRemote(temp, targetFile);
+ }
+ } catch (FileAlreadyExistsException e) {
+ throw new CommitFailedException(
+ "Version %d of table %s was committed by another writer", next, location);
+ } catch (CommitStateUnknownException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeIOException(
+ new java.io.IOException("Unable to commit version " + next + " of " + location, e));
+ }
+ }
+
+ private static void publishLocal(Path temp, Path target) throws java.io.IOException {
+ try {
+ // link(2) fails with EEXIST if the name is taken, atomically.
+ Files.createLink(target, temp);
+ return;
+ } catch (UnsupportedOperationException noHardLinks) {
+ // Fall through to the exclusive create below.
+ } catch (FileAlreadyExistsException | java.nio.file.NoSuchFileException e) {
+ throw e;
+ } catch (FileSystemException e) {
+ if (Files.exists(target)) {
+ throw new FileAlreadyExistsException(target.toString());
+ }
+ // Most likely a file system without hard links: fall through to the exclusive create.
+ }
+ // CREATE_NEW is an exclusive create (O_EXCL): it fails if the name exists, atomically.
+ boolean created = false;
+ try (OutputStream out =
+ Files.newOutputStream(target, java.nio.file.StandardOpenOption.CREATE_NEW)) {
+ created = true;
+ Files.copy(temp, out);
+ } catch (java.io.IOException e) {
+ if (created) {
+ // Never leave a partial version file that readers would take as the current one.
+ Files.deleteIfExists(target);
+ }
+ throw e;
+ }
+ }
+
+ private static void publishRemote(String temp, FileObject target) throws Exception {
+ if (target.exists()) {
+ throw new FileAlreadyExistsException(target.getName().getURI());
+ }
+ try (InputStream in = HopVfs.getInputStream(temp);
+ OutputStream out = HopVfs.getOutputStream(target, false)) {
+ in.transferTo(out);
+ } catch (Exception e) {
+ // The file didn't exist before this write, so a partial one is ours to remove.
+ try {
+ target.delete();
+ } catch (Exception deleteFailed) {
+ throw new CommitStateUnknownException(
+ new java.io.IOException(
+ "Writing " + target.getName().getURI() + " failed and it couldn't be removed", e));
+ }
+ throw e;
+ }
+ }
+
+ private static void deleteQuietly(String file) {
+ try {
+ FileObject object = HopVfs.getFileObject(file);
+ if (object.exists()) {
+ object.delete();
+ }
+ } catch (Exception e) {
+ LogChannel.GENERAL.logBasic(
+ "Unable to delete temporary file " + file + ": " + e.getMessage());
+ }
+ }
+
+ private void writeVersionHint(int newVersion) throws Exception {
+ FileObject hint = HopVfs.getFileObject(metadataFileLocation("version-hint.text"));
+ try (OutputStream out = HopVfs.getOutputStream(hint, false)) {
+ out.write(Integer.toString(newVersion).getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ /**
+ * True when the table's metadata files follow the path-table naming, so a commit here is seen by
+ * every other reader of the folder. Tables created by other catalogs use different names and must
+ * be written through their catalog.
+ */
+ public boolean isPathTable() {
+ TableMetadata metadata = current();
+ if (metadata == null) {
+ return true;
+ }
+ String file = metadata.metadataFileLocation();
+ return file != null && file.substring(file.lastIndexOf('/') + 1).startsWith("v");
+ }
+
+ @Override
+ public FileIO io() {
+ return io;
+ }
+
+ @Override
+ public String metadataFileLocation(String fileName) {
+ return location + "/metadata/" + fileName;
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ TableMetadata metadata = current();
+ return LocationProviders.locationsFor(
+ location, metadata == null ? java.util.Map.of() : metadata.properties());
+ }
+
+ public String location() {
+ return location;
+ }
+}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java
index b2802895e3..f91476801e 100644
--- a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java
@@ -17,17 +17,47 @@
package org.apache.hop.lakehouse.transforms;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.Const;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.lakehouse.LakeFormats;
+import org.apache.hop.lakehouse.iceberg.CommitCoordinator;
+import org.apache.hop.lakehouse.iceberg.CommitCoordinator.WriteMode;
+import org.apache.hop.lakehouse.iceberg.IcebergRowWriter;
+import org.apache.hop.lakehouse.iceberg.IcebergTableTarget;
+import org.apache.hop.lakehouse.iceberg.IcebergTables;
+import org.apache.hop.lakehouse.iceberg.IcebergTypeMapper;
+import org.apache.hop.lakehouse.iceberg.PluginClassLoader;
+import org.apache.hop.lakehouse.metadata.LakeCatalog;
+import org.apache.hop.pipeline.IExecutionFinishedListener;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engine.IPipelineEngine;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
/**
- * Metadata-only on Local engine. On the native Spark engine this becomes a lake table write action.
+ * Writes rows to a lake table on the local engine and passes them on unchanged.
+ *
+ * <p>Every copy of the transform writes its own Parquet files. Nothing becomes visible in the table
+ * until the whole pipeline has finished: then all files of the run are committed as one Iceberg
+ * snapshot, or deleted if the pipeline failed or was stopped. A transform further down the pipeline
+ * that fails after this one has written its rows still rolls the run back, the way a database
+ * transaction would. On the Spark engine this class isn't used.
*/
public class LakeTableOutput extends BaseTransform<LakeTableOutputMeta, LakeTableOutputData> {
+ private static final String COORDINATOR_KEY = "lakehouse.iceberg.coordinator.";
+
public LakeTableOutput(
TransformMeta transformMeta,
LakeTableOutputMeta meta,
@@ -40,7 +70,272 @@ public class LakeTableOutput extends BaseTransform<LakeTableOutputMeta, LakeTabl
@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);
+ }
+ }
+
+ /** Coalesce and extra options configure a Spark write; this engine doesn't use them. */
+ private void warnAboutSparkOnlyOptions() {
+ for (String[] option :
+ new String[][] {
+ {"Coalesce partitions", meta.getCoalescePartitions()},
+ {"Extra options", meta.getExtraOptions()}
+ }) {
+ if (StringUtils.isNotBlank(resolve(option[1]))) {
+ logError(
+ "Warning: '"
+ + option[0]
+ + "' only applies to the Spark engine and is ignored when writing on this engine");
+ }
+ }
+ }
+
+ private CommitCoordinator newCoordinator(IRowMeta rowMeta) throws HopException {
+ warnAboutSparkOnlyOptions();
+ 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();
+ }
+
+ /**
+ * Commits when the pipeline has finished. Hop's own execution-finished listener, which marks the
+ * pipeline finished and stops it running, is put first in the list when the threads start, and
+ * this is called from processRow(), after that. On a local Pipeline the commit listener is
+ * therefore inserted in front of it: the commit runs while the pipeline still reports itself as
+ * running, so a status poll (the remote engine, a parent workflow) can't see "Finished" before
+ * the snapshot exists or before a failed commit is counted.
+ */
+ private void commitWhenPipelineFinishes(CommitCoordinator coordinator) {
+ IExecutionFinishedListener<IPipelineEngine<PipelineMeta>> listener =
+ pipeline -> {
+ // This runs on the thread of whichever transform finished last.
+ try (PluginClassLoader ignored = PluginClassLoader.activate()) {
+ finishRun(pipeline.getErrors() > 0 || pipeline.isStopped(), coordinator);
+ }
+ };
+ if (getPipeline() instanceof Pipeline local) {
+ List<IExecutionFinishedListener<IPipelineEngine<PipelineMeta>>> listeners =
+ local.getExecutionFinishedListeners();
+ synchronized (listeners) {
+ listeners.add(0, listener);
+ }
+ } else {
+ getPipeline().addExecutionFinishedListener(listener);
+ }
+ }
+
+ /**
+ * Commits or aborts the run. A failed commit is counted as an error of this transform before the
+ * listener returns, so it is part of the pipeline result before anyone waiting is released.
+ */
+ private void finishRun(boolean failed, CommitCoordinator coordinator) throws HopException {
+ if (failed) {
+ coordinator.abort();
+ logBasic("The pipeline didn't finish successfully: no rows were committed");
+ return;
+ }
+ try {
+ Long snapshotId = coordinator.commit();
+ logBasic(
+ "Committed "
+ + coordinator.fileCount()
+ + " data file(s)"
+ + (snapshotId == null ? "" : " as snapshot " + snapshotId)
+ + " to table "
+ + coordinator.table().name());
+ } catch (CommitStateUnknownException e) {
+ // The catalog may have applied the commit: deleting the files could break the table.
+ setErrors(getErrors() + 1);
+ throw new HopException(
+ "The commit to Iceberg table "
+ + coordinator.table().name()
+ + " may or may not have been applied. The data files are kept; check the table"
+ + " history before running the pipeline again.",
+ e);
+ } catch (Exception e) {
+ setErrors(getErrors() + 1);
+ if (!coordinator.isPublished()) {
+ coordinator.abort();
+ }
+ throw new HopException("Unable to commit to Iceberg table", e);
+ }
+ }
+
+ @Override
+ public void dispose() {
+ if (data.writer != null) {
+ try (PluginClassLoader ignored = PluginClassLoader.activate()) {
+ data.writer.abort();
+ } catch (HopException e) {
+ logError("Unable to clean up data files", e);
+ }
+ data.writer = null;
+ }
+ super.dispose();
+ }
}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputData.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputData.java
index f5510ddb10..dffa637796 100644
--- a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputData.java
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputData.java
@@ -17,10 +17,19 @@
package org.apache.hop.lakehouse.transforms;
+import org.apache.hop.lakehouse.iceberg.CommitCoordinator;
+import org.apache.hop.lakehouse.iceberg.IcebergRowWriter;
import org.apache.hop.pipeline.transform.BaseTransformData;
import org.apache.hop.pipeline.transform.ITransformData;
public class LakeTableOutputData extends BaseTransformData implements ITransformData {
+
+ /** Shared by all copies of the transform: commits their files once the pipeline is done. */
+ public CommitCoordinator coordinator;
+
+ /** Writes this copy's data files, or null when nothing is written. */
+ public IcebergRowWriter writer;
+
public LakeTableOutputData() {
super();
}
diff --git a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputMeta.java b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputMeta.java
index 02fd1abf6e..f807a3ba7e 100644
--- a/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputMeta.java
+++ b/plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutputMeta.java
@@ -38,7 +38,11 @@ import org.apache.hop.pipeline.transform.TransformMeta;
categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.BigData",
keywords = "i18n::SparkLakeTableOutput.Keyword",
documentationUrl = "/pipeline/transforms/spark-lake-table-output.html",
- supportedEngines = {LakehouseConst.SPARK_ENGINE_ID})
+ supportedEngines = {
+ LakehouseConst.SPARK_ENGINE_ID,
+ LakehouseConst.LOCAL_ENGINE_ID,
+ LakehouseConst.REMOTE_ENGINE_ID
+ })
@Getter
@Setter
public class LakeTableOutputMeta extends BaseTransformMeta<LakeTableOutput, LakeTableOutputData> {
diff --git a/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/CommitCoordinatorTest.java b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/CommitCoordinatorTest.java
new file mode 100644
index 0000000000..bd01a8f72a
--- /dev/null
+++ b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/CommitCoordinatorTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
+import org.apache.iceberg.io.FileIO;
+import org.junit.jupiter.api.Test;
+
+/** The run's data files are only deleted when the table can't be referencing them. */
+class CommitCoordinatorTest {
+
+ private final FileIO io = mock(FileIO.class);
+ private final AppendFiles append = mock(AppendFiles.class);
+ private final Table table = mock(Table.class);
+ private final DataFile file = mock(DataFile.class);
+
+ private CommitCoordinator coordinator() {
+ when(table.io()).thenReturn(io);
+ when(table.newAppend()).thenReturn(append);
+ when(file.location()).thenReturn("file:///lake/orders/data/a.parquet");
+ CommitCoordinator coordinator =
+ new CommitCoordinator(table, CommitCoordinator.WriteMode.APPEND, Map.of());
+ coordinator.addFiles(List.of(file));
+ return coordinator;
+ }
+
+ @Test
+ void unknownOutcomeKeepsTheFiles() {
+ CommitCoordinator coordinator = coordinator();
+ doThrow(new CommitStateUnknownException(new RuntimeException("timeout"))).when(append).commit();
+
+ assertThrows(CommitStateUnknownException.class, coordinator::commit);
+
+ assertTrue(coordinator.isOutcomeUnknown());
+ assertFalse(coordinator.isPublished());
+ assertThrows(IllegalStateException.class, coordinator::abort);
+ verify(io, never()).deleteFile(anyString());
+ }
+
+ @Test
+ void failureAfterPublishingKeepsTheFiles() {
+ CommitCoordinator coordinator = coordinator();
+ doThrow(new RuntimeException("refresh failed")).when(table).refresh();
+
+ // The snapshot is in the table; only reading it back failed.
+ coordinator.commit();
+
+ assertTrue(coordinator.isPublished());
+ assertThrows(IllegalStateException.class, coordinator::abort);
+ verify(io, never()).deleteFile(anyString());
+ }
+
+ @Test
+ void definiteFailureDeletesTheFiles() {
+ CommitCoordinator coordinator = coordinator();
+ doThrow(new CommitFailedException("conflict")).when(append).commit();
+
+ assertThrows(CommitFailedException.class, coordinator::commit);
+
+ assertFalse(coordinator.isPublished());
+ coordinator.abort();
+ verify(io).deleteFile("file:///lake/orders/data/a.parquet");
+ }
+}
diff --git a/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/IcebergTablesTest.java b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/IcebergTablesTest.java
index a4bccf4a13..900550c800 100644
--- a/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/IcebergTablesTest.java
+++ b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/IcebergTablesTest.java
@@ -145,16 +145,26 @@ class IcebergTablesTest {
assertEquals(-1, IcebergTables.metadataVersion("snap-123-1-abc.avro"));
}
+ /**
+ * The hint is a starting point: a writer publishes {@code v<N>.metadata.json} before it updates
+ * the hint, so newer versions after the hinted one are followed, like Iceberg's Hadoop tables do,
+ * including both gzip spellings of the metadata file name. The walk stops at the first missing
+ * version.
+ */
@Test
- void versionHintWins() throws Exception {
+ void versionHintIsAStartingPoint() throws Exception {
Path metadata = Files.createDirectories(tempDir.resolve("t/metadata"));
Files.writeString(metadata.resolve("v1.metadata.json"), "{}");
Files.writeString(metadata.resolve("v2.metadata.json"), "{}");
+ Files.writeString(metadata.resolve("v3.gz.metadata.json"), "{}");
+ Files.writeString(metadata.resolve("v4.metadata.json.gz"), "{}");
+ Files.writeString(metadata.resolve("v6.metadata.json"), "{}");
Files.writeString(metadata.resolve("version-hint.text"), "1\n");
String root = tempDir.resolve("t").toUri().toString();
assertEquals(
- metadataFolder(root) + "/v1.metadata.json", IcebergTables.currentMetadataFile(root));
+ metadataFolder(root) + "/v4.metadata.json.gz", IcebergTables.currentMetadataFile(root));
+ assertEquals(4, IcebergTables.metadataVersion("v4.metadata.json.gz"));
}
@Test
diff --git a/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/PathTableOperationsTest.java b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/PathTableOperationsTest.java
new file mode 100644
index 0000000000..5a006c8752
--- /dev/null
+++ b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/iceberg/PathTableOperationsTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class PathTableOperationsTest {
+
+ private static final Schema SCHEMA =
+ new Schema(Types.NestedField.required(1, "id", Types.LongType.get()));
+
+ @TempDir Path tempDir;
+
+ private Path tableDir;
+ private String location;
+ private final AtomicInteger fileCounter = new AtomicInteger();
+
+ @BeforeAll
+ static void init() throws Exception {
+ HopEnvironment.init();
+ }
+
+ @BeforeEach
+ void createTable() throws Exception {
+ tableDir = tempDir.resolve("orders");
+ location = tableDir.toUri().toString();
+ IcebergTableTarget.atPath(location)
+ .create(SCHEMA, PartitionSpec.unpartitioned(), Map.of())
+ .commitTransaction();
+ }
+
+ private Path metadata(String name) {
+ return tableDir.resolve("metadata").resolve(name);
+ }
+
+ private DataFile dataFile() {
+ return DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(
+ tableDir.resolve("data/file-" + fileCounter.incrementAndGet() + ".parquet").toString())
+ .withFormat(FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .build();
+ }
+
+ private void append() {
+ PathTableOperations ops = new PathTableOperations(location);
+ new BaseTable(ops, location).newAppend().appendFile(dataFile()).commit();
+ }
+
+ private void setHint(String text) throws Exception {
+ Files.writeString(metadata("version-hint.text"), text, StandardCharsets.UTF_8);
+ }
+
+ @Test
+ void staleHintIsFollowedForward() throws Exception {
+ append();
+ append();
+ assertTrue(Files.exists(metadata("v3.metadata.json")));
+ // A crash between publishing v3 and updating the hint leaves the hint behind.
+ setHint("1");
+
+ assertTrue(
+ IcebergTables.findCurrentMetadataFile(location).endsWith("/metadata/v3.metadata.json"));
+ assertEquals(2, snapshotCount(IcebergTables.loadFromPath(location)));
+
+ append();
+ assertTrue(Files.exists(metadata("v4.metadata.json")), "the next commit writes v4");
+ assertEquals("4", Files.readString(metadata("version-hint.text")).trim());
+ }
+
+ @Test
+ void unreadableHintFallsBackToTheNewestFile() throws Exception {
+ append();
+ Files.delete(metadata("version-hint.text"));
+ Files.createDirectory(metadata("version-hint.text"));
+
+ append();
+
+ assertTrue(Files.exists(metadata("v3.metadata.json")), "the commit succeeds without a hint");
+ assertTrue(
+ IcebergTables.findCurrentMetadataFile(location).endsWith("/metadata/v3.metadata.json"));
+ }
+
+ @Test
+ void publishRefusesAVersionThatIsTaken() throws Exception {
+ Path v1 = metadata("v1.metadata.json");
+ String before = Files.readString(v1);
+ Path temp = metadata("other-writer.metadata.json.tmp");
+ Files.writeString(temp, "{}", StandardCharsets.UTF_8);
+
+ PathTableOperations ops = new PathTableOperations(location);
+ assertThrows(
+ CommitFailedException.class,
+ () -> ops.publish(temp.toUri().toString(), v1.toUri().toString(), 1));
+ assertEquals(before, Files.readString(v1), "the committed version is left alone");
+ }
+
+ @Test
+ void concurrentWritersInOneJvmLoseNoSnapshot() throws Exception {
+ int writers = 8;
+ int appendsEach = 5;
+ ExecutorService pool = Executors.newFixedThreadPool(writers);
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<Integer>> results = new ArrayList<>();
+ for (int w = 0; w < writers; w++) {
+ Callable<Integer> writer =
+ () -> {
+ start.await();
+ int failed = 0;
+ for (int i = 0; i < appendsEach; i++) {
+ try {
+ append();
+ } catch (CommitFailedException e) {
+ failed++;
+ }
+ }
+ return failed;
+ };
+ results.add(pool.submit(writer));
+ }
+ start.countDown();
+ int failed = 0;
+ for (Future<Integer> result : results) {
+ failed += result.get();
+ }
+ pool.shutdown();
+
+ Table table = IcebergTables.loadFromPath(location);
+ int committed = writers * appendsEach - failed;
+ assertEquals(
+ committed, snapshotCount(table), "every commit that reported success is in the table");
+ assertEquals(
+ String.valueOf(committed),
+ table.currentSnapshot().summary().get("total-data-files"),
+ "and every appended file is still referenced");
+ }
+
+ private static int snapshotCount(Table table) {
+ int count = 0;
+ for (var ignored : table.snapshots()) {
+ count++;
+ }
+ return count;
+ }
+
+ /**
+ * On a file system without hard links, here Hop VFS's in-memory ram://, a version file that
+ * already exists is never deleted or replaced: the commit fails instead.
+ */
+ @Test
+ void nonLocalPublishNeverReplacesAVersion() throws Exception {
+ String ram = "ram:///path-table-test-" + java.util.UUID.randomUUID();
+ String target = ram + "/metadata/v2.metadata.json";
+ String temp = ram + "/metadata/other.metadata.json.tmp";
+ write(target, "committed by the first writer");
+ write(temp, "second writer");
+
+ PathTableOperations ops = new PathTableOperations(ram);
+ assertThrows(CommitFailedException.class, () -> ops.publish(temp, target, 2));
+ assertEquals("committed by the first writer", read(target));
+
+ String free = ram + "/metadata/v3.metadata.json";
+ ops.publish(temp, free, 3);
+ assertEquals("second writer", read(free));
+ }
+
+ @Test
+ void pathTableOnANonLocalFileSystem() throws Exception {
+ String ram = "ram:///path-table-test-" + java.util.UUID.randomUUID();
+ IcebergTableTarget.atPath(ram)
+ .create(SCHEMA, PartitionSpec.unpartitioned(), Map.of())
+ .commitTransaction();
+ PathTableOperations ops = new PathTableOperations(ram);
+ new BaseTable(ops, ram).newAppend().appendFile(dataFile()).commit();
+
+ assertEquals(1, snapshotCount(IcebergTables.loadFromPath(ram)));
+ assertTrue(IcebergTables.findCurrentMetadataFile(ram).endsWith("/metadata/v2.metadata.json"));
+ }
+
+ private static void write(String uri, String text) throws Exception {
+ org.apache.commons.vfs2.FileObject file = org.apache.hop.core.vfs.HopVfs.getFileObject(uri);
+ file.getParent().createFolder();
+ try (java.io.OutputStream out = org.apache.hop.core.vfs.HopVfs.getOutputStream(file, false)) {
+ out.write(text.getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ private static String read(String uri) throws Exception {
+ try (java.io.InputStream in = org.apache.hop.core.vfs.HopVfs.getInputStream(uri)) {
+ return new String(in.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ }
+
+ /** On a local disk the version is published as a hard link of the temp file, atomically. */
+ @Test
+ void localPublishUsesAHardLink() throws Exception {
+ org.junit.jupiter.api.Assumptions.assumeTrue(
+ java.nio.file.FileSystems.getDefault().supportedFileAttributeViews().contains("unix"));
+ Path temp = metadata("next.metadata.json.tmp");
+ Path target = metadata("v7.metadata.json");
+ Files.writeString(temp, "{}", StandardCharsets.UTF_8);
+
+ new PathTableOperations(location)
+ .publish(temp.toUri().toString(), target.toUri().toString(), 7);
+
+ assertEquals(2, Files.getAttribute(target, "unix:nlink"), "target and temp are one file");
+ }
+}
diff --git a/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/transforms/LakeTableOutputLocalTest.java b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/transforms/LakeTableOutputLocalTest.java
new file mode 100644
index 0000000000..7d584ce9ee
--- /dev/null
+++ b/plugins/tech/lakehouse/src/test/java/org/apache/hop/lakehouse/transforms/LakeTableOutputLocalTest.java
@@ -0,0 +1,554 @@
+/*
+ * 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.transforms;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBigNumber;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.row.value.ValueMetaTimestamp;
+import org.apache.hop.lakehouse.LakeFormats;
+import org.apache.hop.lakehouse.iceberg.CommitCoordinator;
+import org.apache.hop.lakehouse.iceberg.IcebergRowReader;
+import org.apache.hop.lakehouse.iceberg.IcebergTables;
+import org.apache.hop.lakehouse.iceberg.io.HopVfsFileIO;
+import org.apache.hop.lakehouse.metadata.LakeCatalog;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.apache.hop.pipeline.PipelineHopMeta;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.RowProducer;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.RowAdapter;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.dummy.DummyMeta;
+import org.apache.hop.pipeline.transforms.injector.InjectorMeta;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.jdbc.JdbcCatalog;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/** Runs Lake table output on the local engine and checks the Iceberg table it leaves behind. */
+class LakeTableOutputLocalTest {
+
+ private static final String[] REGIONS = {"EU", "US", "APAC"};
+
+ @TempDir Path tempDir;
+
+ private IRowMeta rowMeta;
+ private MemoryMetadataProvider metadataProvider;
+ private String tablePath;
+
+ /** Runs in the pipeline when the first row has been written, before the run commits. */
+ private Runnable onFirstRow;
+
+ @BeforeAll
+ static void init() throws Exception {
+ HopEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() {
+ rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaInteger("id"));
+ rowMeta.addValueMeta(new ValueMetaString("region"));
+ rowMeta.addValueMeta(new ValueMetaBigNumber("amount", 12, 2));
+ metadataProvider = new MemoryMetadataProvider();
+ tablePath = tempDir.resolve("orders").toUri().toString();
+ }
+
+ private static List<Object[]> rows(long from, long to) {
+ List<Object[]> rows = new ArrayList<>();
+ for (long id = from; id < to; id++) {
+ rows.add(new Object[] {id, REGIONS[(int) (id % REGIONS.length)], BigDecimal.valueOf(id, 2)});
+ }
+ return rows;
+ }
+
+ private LakeTableOutputMeta output(String saveMode) {
+ LakeTableOutputMeta meta = new LakeTableOutputMeta();
+ meta.setFormat(LakeFormats.FORMAT_ICEBERG);
+ meta.setIdentifierMode(LakeTableInputMeta.MODE_PATH);
+ meta.setTablePath(tablePath);
+ meta.setSaveMode(saveMode);
+ return meta;
+ }
+
+ private static class Run {
+ int errors;
+ final AtomicInteger passedThrough = new AtomicInteger();
+ }
+
+ /** Injector -> output (copies) -> dummy. Optionally the dummy fails on the last row. */
+ private Run run(LakeTableOutputMeta output, int copies, List<Object[]> rows, boolean failAfter)
+ throws HopException {
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("write-orders");
+ TransformMeta injector = new TransformMeta("injector", new InjectorMeta());
+ TransformMeta writer = new TransformMeta("output", output);
+ writer.setCopies(copies);
+ TransformMeta dummy = new TransformMeta("dummy", new DummyMeta());
+ pipelineMeta.addTransform(injector);
+ pipelineMeta.addTransform(writer);
+ pipelineMeta.addTransform(dummy);
+ pipelineMeta.addPipelineHop(new PipelineHopMeta(injector, writer));
+ pipelineMeta.addPipelineHop(new PipelineHopMeta(writer, dummy));
+
+ LocalPipelineEngine pipeline = new LocalPipelineEngine(pipelineMeta);
+ pipeline.setMetadataProvider(metadataProvider);
+ pipeline.prepareExecution();
+
+ Run run = new Run();
+ pipeline
+ .getTransform("dummy", 0)
+ .addRowListener(
+ new RowAdapter() {
+ @Override
+ public void rowReadEvent(IRowMeta meta, Object[] row) throws HopTransformException {
+ int seen = run.passedThrough.incrementAndGet();
+ if (seen == 1 && onFirstRow != null) {
+ onFirstRow.run();
+ }
+ if (seen == rows.size() && failAfter) {
+ throw new HopTransformException("Failing after the output finished writing");
+ }
+ }
+ });
+ RowProducer producer = pipeline.addRowProducer("injector", 0);
+ pipeline.startThreads();
+ for (Object[] row : rows) {
+ producer.putRow(rowMeta, row.clone());
+ }
+ producer.finished();
+ pipeline.waitUntilFinished();
+ run.errors = pipeline.getErrors();
+ return run;
+ }
+
+ private Table table() throws HopException {
+ return IcebergTables.loadFromPath(tablePath);
+ }
+
+ private static Set<Long> ids(Table table) {
+ Set<Long> ids = new HashSet<>();
+ List<Long> all = new ArrayList<>();
+ try (IcebergRowReader reader = new IcebergRowReader(table, List.of("id"), null, null, 0, 1)) {
+ reader.forEachRemaining(row -> all.add((Long) row[0]));
+ }
+ ids.addAll(all);
+ assertEquals(all.size(), ids.size(), "no row is stored twice");
+ return ids;
+ }
+
+ private long dataFileCount() throws IOException {
+ Path data = tempDir.resolve("orders/data");
+ if (!Files.exists(data)) {
+ return 0;
+ }
+ try (Stream<Path> files = Files.walk(data)) {
+ return files.filter(p -> p.toString().endsWith(".parquet")).count();
+ }
+ }
+
+ @Test
+ void createsTableInOneSnapshotFromAllCopies() throws Exception {
+ Run run = run(output(LakeTableOutputMeta.MODE_ERROR), 3, rows(0, 1000), false);
+
+ assertEquals(0, run.errors);
+ assertEquals(1000, run.passedThrough.get(), "rows are passed on");
+ Table table = table();
+ assertEquals(1, table.history().size(), "one snapshot for the whole run");
+ assertEquals(1000, ids(table).size());
+ assertEquals("write-orders", table.currentSnapshot().summary().get("hop.pipeline"));
+ assertTrue(table.currentSnapshot().summary().containsKey(CommitCoordinator.SUMMARY_RUN_ID));
+ assertTrue(dataFileCount() >= 3, "each copy writes its own files");
+
+ // The Spark engine's path tables use the same layout, so both engines see the same table.
+ assertEquals(
+ "1", Files.readString(tempDir.resolve("orders/metadata/version-hint.text")).trim());
+ assertTrue(Files.exists(tempDir.resolve("orders/metadata/v1.metadata.json")));
+ }
+
+ @Test
+ void typesOfCreatedTable() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 10), false);
+
+ Schema schema = table().schema();
+ assertEquals(Types.LongType.get(), schema.findType("id"));
+ assertEquals(Types.StringType.get(), schema.findType("region"));
+ assertEquals(Types.DecimalType.of(12, 2), schema.findType("amount"));
+ }
+
+ @Test
+ void partitionsByColumns() throws Exception {
+ LakeTableOutputMeta meta = output(LakeTableOutputMeta.MODE_ERROR);
+ meta.setPartitionByColumns("region");
+
+ assertEquals(0, run(meta, 2, rows(0, 300), false).errors);
+
+ Table table = table();
+ assertEquals("region", table.spec().fields().get(0).name());
+ assertEquals(300, ids(table).size());
+ }
+
+ @Test
+ void unknownPartitionColumnFails() throws Exception {
+ LakeTableOutputMeta meta = output(LakeTableOutputMeta.MODE_ERROR);
+ meta.setPartitionByColumns("country");
+
+ assertTrue(run(meta, 1, rows(0, 10), false).errors > 0);
+ assertFalse(Files.exists(tempDir.resolve("orders/metadata")));
+ }
+
+ @Test
+ void appendAddsASnapshot() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 100), false);
+ Run run = run(output(LakeTableOutputMeta.MODE_APPEND), 2, rows(100, 150), false);
+
+ assertEquals(0, run.errors);
+ Table table = table();
+ assertEquals(2, table.history().size());
+ assertEquals(150, ids(table).size());
+ }
+
+ @Test
+ void appendCreatesAMissingTable() throws Exception {
+ assertEquals(0, run(output(LakeTableOutputMeta.MODE_APPEND), 1, rows(0, 20), false).errors);
+ assertEquals(20, ids(table()).size());
+ }
+
+ @Test
+ void overwriteReplacesTheRows() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 100), false);
+ Run run = run(output(LakeTableOutputMeta.MODE_OVERWRITE), 2, rows(500, 530), false);
+
+ assertEquals(0, run.errors);
+ Set<Long> ids = ids(table());
+ assertEquals(30, ids.size());
+ assertTrue(ids.contains(500L) && !ids.contains(0L));
+ }
+
+ @Test
+ void errorIfExistsLeavesAnExistingTableAlone() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 100), false);
+ Run run = run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(100, 200), false);
+
+ assertTrue(run.errors > 0);
+ assertEquals(100, ids(table()).size());
+ }
+
+ @Test
+ void ignorePassesRowsWithoutWriting() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 100), false);
+ Run run = run(output(LakeTableOutputMeta.MODE_IGNORE), 2, rows(100, 200), false);
+
+ assertEquals(0, run.errors);
+ assertEquals(100, run.passedThrough.get());
+ assertEquals(1, table().history().size());
+ assertEquals(100, ids(table()).size());
+ }
+
+ @Test
+ void failureLaterInThePipelineCreatesNoTable() throws Exception {
+ Run run = run(output(LakeTableOutputMeta.MODE_ERROR), 3, rows(0, 500), true);
+
+ assertTrue(run.errors > 0);
+ assertFalse(Files.exists(tempDir.resolve("orders/metadata")), "no table was created");
+ assertEquals(0, dataFileCount(), "the data files written by the run are removed");
+ }
+
+ @Test
+ void failureLaterInThePipelineLeavesTheTableUnchanged() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 100), false);
+ long filesBefore = dataFileCount();
+
+ Run run = run(output(LakeTableOutputMeta.MODE_APPEND), 2, rows(100, 400), true);
+
+ assertTrue(run.errors > 0);
+ Table table = table();
+ assertEquals(1, table.history().size());
+ assertEquals(100, ids(table).size());
+ assertEquals(filesBefore, dataFileCount());
+ }
+
+ @Test
+ void writesThroughAHadoopCatalog() throws Exception {
+ LakeCatalog lake = new LakeCatalog();
+ lake.setName("lake");
+ lake.setCatalogName("lake");
+ lake.setCatalogType(LakeCatalog.TYPE_HADOOP);
+ lake.setWarehouse(tempDir.resolve("warehouse").toUri().toString());
+ metadataProvider.getSerializer(LakeCatalog.class).save(lake);
+
+ LakeTableOutputMeta meta = new LakeTableOutputMeta();
+ meta.setFormat(LakeFormats.FORMAT_ICEBERG);
+ meta.setIdentifierMode(LakeTableInputMeta.MODE_TABLE);
+ meta.setCatalogMetadataName("lake");
+ meta.setTableIdentifier("lake.sales.orders");
+ meta.setSaveMode(LakeTableOutputMeta.MODE_APPEND);
+
+ assertEquals(0, run(meta, 2, rows(0, 120), false).errors);
+ Table table =
+ IcebergTables.loadFromPath(tempDir.resolve("warehouse/sales/orders").toUri().toString());
+ assertEquals(120, ids(table).size());
+ }
+
+ @Test
+ void refusesToWriteToACatalogTableByPath() throws Exception {
+ try (JdbcCatalog catalog = new JdbcCatalog()) {
+ catalog.initialize(
+ "lake",
+ Map.of(
+ CatalogProperties.URI,
+ "jdbc:sqlite:" + tempDir.resolve("catalog.db"),
+ CatalogProperties.WAREHOUSE_LOCATION,
+ tempDir.resolve("warehouse").toUri().toString(),
+ CatalogProperties.FILE_IO_IMPL,
+ HopVfsFileIO.class.getName(),
+ "jdbc.schema-version",
+ "V1"));
+ catalog.createNamespace(Namespace.of("sales"));
+ Table managed =
+ catalog.createTable(
+ TableIdentifier.of("sales", "orders"),
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())));
+ tablePath = managed.location();
+
+ Run run = run(output(LakeTableOutputMeta.MODE_APPEND), 1, rows(0, 10), false);
+
+ assertTrue(run.errors > 0);
+ managed.refresh();
+ assertEquals(null, managed.currentSnapshot());
+ }
+ }
+
+ @Test
+ void deltaIsNotSupportedOnTheLocalEngine() throws Exception {
+ LakeTableOutputMeta meta = output(LakeTableOutputMeta.MODE_ERROR);
+ meta.setFormat(LakeFormats.FORMAT_DELTA);
+
+ assertTrue(run(meta, 1, rows(0, 10), false).errors > 0);
+ }
+
+ /**
+ * A Hop timestamp is a local date and time, so it is stored as an Iceberg timestamp without zone
+ * holding the same wall-clock time, whatever the JVM time zone is, and reads back unchanged.
+ */
+ @Test
+ void timestampsKeepTheirWallClockTimeOutsideUtc() throws Exception {
+ TimeZone original = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"));
+ rowMeta.addValueMeta(new ValueMetaTimestamp("ts"));
+ LocalDateTime noon = LocalDateTime.of(2026, 10, 1, 12, 0);
+ List<Object[]> rows = new ArrayList<>();
+ for (Object[] row : rows(0, 3)) {
+ rows.add(new Object[] {row[0], row[1], row[2], Timestamp.valueOf(noon)});
+ }
+
+ Run run = run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows, false);
+
+ assertEquals(0, run.errors);
+ Table table = table();
+ assertEquals(Types.TimestampType.withoutZone(), table.schema().findType("ts"));
+ try (CloseableIterable<Record> records = IcebergGenerics.read(table).build()) {
+ for (Record record : records) {
+ assertEquals(noon, record.getField("ts"), "stored as the same wall-clock time");
+ }
+ }
+ try (IcebergRowReader reader = new IcebergRowReader(table, List.of("ts"), null, null, 0, 1)) {
+ assertEquals(noon, ((Timestamp) reader.next()[0]).toLocalDateTime());
+ }
+ } finally {
+ TimeZone.setDefault(original);
+ }
+ }
+
+ @Test
+ void unknownSaveModeIsRefused() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 10), false);
+
+ Run run = run(output("Upsert"), 1, rows(10, 20), false);
+
+ assertTrue(run.errors > 0, "an unsupported save mode fails instead of appending");
+ assertEquals(10, ids(table()).size(), "the existing table is unchanged");
+ }
+
+ @Test
+ void unknownSaveModeCreatesNoTable() throws Exception {
+ Run run = run(output("Upsert"), 1, rows(0, 10), false);
+
+ assertTrue(run.errors > 0);
+ assertFalse(Files.exists(tempDir.resolve("orders/metadata")), "no table is created");
+ }
+
+ /**
+ * A commit that fails is part of the pipeline result as soon as waitUntilFinished() returns, so a
+ * parent workflow can't treat the run as successful. Here an overwrite fails validation because
+ * another writer appended to the table while the run was writing.
+ */
+ @Test
+ void failedCommitIsAnErrorWhenThePipelineFinishes() throws Exception {
+ run(output(LakeTableOutputMeta.MODE_ERROR), 1, rows(0, 10), false);
+ onFirstRow =
+ () -> {
+ Table other =
+ new org.apache.iceberg.BaseTable(
+ new org.apache.hop.lakehouse.iceberg.PathTableOperations(tablePath), tablePath);
+ other
+ .newAppend()
+ .appendFile(
+ org.apache.iceberg.DataFiles.builder(other.spec())
+ .withPath(tablePath + "/data/concurrent.parquet")
+ .withFormat(org.apache.iceberg.FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .build())
+ .commit();
+ };
+
+ Run run = run(output(LakeTableOutputMeta.MODE_OVERWRITE), 1, rows(100, 120), false);
+
+ assertTrue(run.errors > 0, "the failed commit is in the result waitUntilFinished() returns");
+ assertEquals(
+ 2, table().history().size(), "the overwrite didn't replace the other writer's data");
+ }
+
+ /**
+ * An existing table without snapshots is protected too: a writer that gives it its first snapshot
+ * while an Overwrite runs makes the overwrite fail instead of being removed by it.
+ */
+ @Test
+ void overwriteOfAnEmptyTableDoesntRemoveAConcurrentFirstAppend() throws Exception {
+ org.apache.hop.lakehouse.iceberg.PathTableOperations ops =
+ new org.apache.hop.lakehouse.iceberg.PathTableOperations(tablePath);
+ org.apache.iceberg.Transactions.createTableTransaction(
+ tablePath,
+ ops,
+ org.apache.iceberg.TableMetadata.newTableMetadata(
+ org.apache.hop.lakehouse.iceberg.IcebergTypeMapper.toIcebergSchema(rowMeta),
+ org.apache.iceberg.PartitionSpec.unpartitioned(),
+ tablePath,
+ Map.of()))
+ .commitTransaction();
+ assertTrue(table().currentSnapshot() == null, "the table starts without a snapshot");
+ onFirstRow =
+ () -> {
+ Table other =
+ new org.apache.iceberg.BaseTable(
+ new org.apache.hop.lakehouse.iceberg.PathTableOperations(tablePath), tablePath);
+ other
+ .newAppend()
+ .appendFile(
+ org.apache.iceberg.DataFiles.builder(other.spec())
+ .withPath(tablePath + "/data/first.parquet")
+ .withFormat(org.apache.iceberg.FileFormat.PARQUET)
+ .withFileSizeInBytes(10)
+ .withRecordCount(1)
+ .build())
+ .commit();
+ };
+
+ Run run = run(output(LakeTableOutputMeta.MODE_OVERWRITE), 1, rows(0, 20), false);
+
+ assertTrue(run.errors > 0, "the overwrite fails validation");
+ assertEquals(1, table().history().size(), "the other writer's first snapshot is kept");
+ assertEquals(
+ tablePath + "/data/first.parquet",
+ table().currentSnapshot().addedDataFiles(table().io()).iterator().next().location());
+ }
+
+ /**
+ * The commit runs while the pipeline still reports itself as running: whoever polls the status
+ * never sees "Finished" before the snapshot is in the table.
+ */
+ @Test
+ void statusIsNotFinishedBeforeTheSnapshotExists() throws Exception {
+ java.util.concurrent.atomic.AtomicReference<String> seenWhenFinished =
+ new java.util.concurrent.atomic.AtomicReference<>();
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.setName("write-orders");
+ TransformMeta injector = new TransformMeta("injector", new InjectorMeta());
+ TransformMeta writer = new TransformMeta("output", output(LakeTableOutputMeta.MODE_ERROR));
+ writer.setCopies(3);
+ pipelineMeta.addTransform(injector);
+ pipelineMeta.addTransform(writer);
+ pipelineMeta.addPipelineHop(new PipelineHopMeta(injector, writer));
+ LocalPipelineEngine pipeline = new LocalPipelineEngine(pipelineMeta);
+ pipeline.setMetadataProvider(metadataProvider);
+ pipeline.prepareExecution();
+ RowProducer producer = pipeline.addRowProducer("injector", 0);
+ pipeline.startThreads();
+
+ Thread poller =
+ new Thread(
+ () -> {
+ while (seenWhenFinished.get() == null) {
+ if (pipeline
+ .getStatus()
+ .startsWith(org.apache.hop.pipeline.Pipeline.STRING_FINISHED)) {
+ boolean exists =
+ java.nio.file.Files.exists(
+ tempDir.resolve("orders/metadata/v1.metadata.json"));
+ seenWhenFinished.set(exists ? "snapshot" : "no snapshot");
+ }
+ }
+ });
+ poller.start();
+ for (Object[] row : rows(0, 3000)) {
+ producer.putRow(rowMeta, row.clone());
+ }
+ producer.finished();
+ pipeline.waitUntilFinished();
+ poller.join(10_000);
+
+ assertEquals(0, pipeline.getErrors());
+ assertEquals("snapshot", seenWhenFinished.get());
+ }
+}