mattcasters commented on code in PR #8819: URL: https://github.com/apache/hop/pull/8819#discussion_r4229707629
########## plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/PathTableOperations.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.util.UUID; +import org.apache.commons.vfs2.FileObject; +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 moves it to the next version's name, + * failing if that name is already taken. On a local or HDFS file system the move is atomic, so two + * writers can't both commit the same version. Object stores like S3 can't guarantee that, the same + * limitation Iceberg documents for Hadoop tables, so there a table should have one writer at a + * time. Tables managed by a catalog should be written through the catalog instead. + */ +public class PathTableOperations implements TableOperations { + + 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) { + 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 { + FileObject tempFile = HopVfs.getFileObject(temp); + FileObject targetFile = HopVfs.getFileObject(target); + if (targetFile.exists()) { + tempFile.delete(); + throw new CommitFailedException( + "Version %d of table %s was committed by another writer", next, location); + } + tempFile.moveTo(targetFile); Review Comment: **[bug]** `moveTo` does not fail when `v<N>.metadata.json` is already taken, so two path-table writers can both commit the same version. Commons VFS `FileObject.moveTo` deletes the destination first if it exists, and a local `rename` overwrites. The `exists` check above is not atomic with the move, so both writers can pass it, and the later `moveTo` deletes the earlier metadata file. Both callers then see a successful commit, but only one snapshot remains. That contradicts the class comment that a local or HDFS move makes a conflicting commit fail. **Suggestion:** Publish with an exclusive create that fails if the target name exists (local `CREATE_NEW` / `Files.move` without `REPLACE_EXISTING`, or HDFS rename that does not overwrite). Do not use VFS `moveTo` for this. Delete the temp metadata file in a `finally` when the publish does not succeed. Object stores still cannot do this atomically; that single-writer limitation should stay. ########## plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java: ########## @@ -40,7 +67,194 @@ public LakeTableOutput( @Override public boolean processRow() throws HopException { - setOutputDone(); - return false; + 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 = + LakeTableOutputMeta.MODE_OVERWRITE.equalsIgnoreCase(saveMode) + ? WriteMode.OVERWRITE_TABLE + : WriteMode.APPEND; Review Comment: **[suggestion]** An existing table with any save mode other than Overwrite, ErrorIfExists, or Ignore is appended to. A typo, extra whitespace, or an injected value such as `Upsert` therefore writes rows instead of failing. The Spark writer rejects an unsupported mode. **Suggestion:** Accept only `Append` for append. Throw `HopException` for anything else, after the ErrorIfExists and Ignore handling above. ########## plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/iceberg/IcebergTables.java: ########## @@ -189,27 +202,38 @@ static Map<String, String> extraProperties(String confExtra) { 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. + * The version hint is followed when there is one; otherwise the newest metadata file 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()) { 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"; + String file = metadataFolder + "/v" + version + ".metadata.json"; + if (HopVfs.getFileObject(file).exists()) { + return file; Review Comment: **[bug]** A version hint that still names an older existing file is returned immediately, so a newer `v<N>.metadata.json` is invisible. `PathTableOperations` now allocates the next version from this method, and it writes the metadata file before `version-hint.text`. Iceberg's `HadoopTableOperations.refresh` walks `v{hint+1}`, `v{hint+2}`, ... because the hint is best-effort and is often stale after a crash between those two writes. Here the hint wins whenever that older file still exists, which it does. The next commit then tries the same version, sees the file, and fails forever. Spark, which scans forward, reads the newer snapshot while this engine stays on the old one, so the two engines no longer see the same table. **Suggestion:** After the hinted file, keep walking newer `v<N>.metadata.json` and `v<N>.gz.metadata.json` files and use the highest that exists. Treat a failed hint update as non-fatal once the metadata file is in place, matching `HadoopTableOperations.writeVersionHint`. Add a test that leaves `v2.metadata.json` on disk with the hint still at `1` and checks that readers see v2 and the next commit writes v3. ########## plugins/tech/lakehouse/src/main/java/org/apache/hop/lakehouse/transforms/LakeTableOutput.java: ########## @@ -40,7 +67,194 @@ public LakeTableOutput( @Override public boolean processRow() throws HopException { - setOutputDone(); - return false; + 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 = + LakeTableOutputMeta.MODE_OVERWRITE.equalsIgnoreCase(saveMode) + ? WriteMode.OVERWRITE_TABLE + : WriteMode.APPEND; + return new CommitCoordinator(target.load(), mode, summary); + } + + 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 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( + pipeline -> { + // This runs on the thread of whichever transform finished last. + try (PluginClassLoader ignored = PluginClassLoader.activate()) { + finishRun(pipeline.getErrors() > 0 || pipeline.isStopped(), coordinator); + } + }); + } + + 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) as snapshot " + + snapshotId + + " of table " + + coordinator.table().name()); + } catch (Exception e) { + coordinator.abort(); Review Comment: **[bug]** Any exception from `coordinator.commit()` deletes the run's data files, including when the snapshot metadata is already published. `CommitCoordinator.commit` calls `update.commit()` and, for a new table, `commitTransaction()` before `table.refresh()` and `currentSnapshot()`. `PathTableOperations.commit` has already moved `v<N>.metadata.json` into place before `writeVersionHint`. A hint-write failure, or a `refresh()` failure after a successful append, is caught here and `abort()` removes the Parquet files that metadata now references. Local readers are then stuck or see a snapshot with missing files, and Spark (which discovers the new metadata file even with a stale hint) reads a broken snapshot. **Suggestion:** Remember whether the metadata publish succeeded and only `abort()` when it did not. Once `update.commit()` or `commitTransaction()` returns, do not delete data files; log a refresh failure instead of failing the run. Inside `PathTableOperations.commit`, do not fail the commit if updating `version-hint.text` throws after the metadata file is in place. -- 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]
