mattcasters commented on code in PR #8859:
URL: https://github.com/apache/hop/pull/8859#discussion_r4243062395
##########
docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-lake-table-input.adoc:
##########
@@ -92,6 +97,24 @@ This transform is **not** available on the local Hop engine
or on Beam engines.
Tip: After two Overwrite writes, Delta version `0` is typically the first
commit. For Iceberg, inspect `+{table}.snapshots+` (or catalog metadata) for
snapshot ids.
+[[hop-engine]]
+== On the Hop engine
+
+On the local and remote Hop engines the transform reads Apache Iceberg tables.
Delta Lake tables give an error that asks you to use the Spark engine for now.
+
+*Finding the table*
+
+* *PATH* mode reads the table in the given folder. The folder can be any Hop
VFS location, for example `file:///data/orders` or `s3://lake/orders`. When the
table has a `metadata/version-hint.text` file (tables written by a Hadoop
catalog, by the Spark engine in PATH mode, or by Lake table output on the Hop
engine) the version it names is read. Otherwise the newest metadata file is
read.
+* *TABLE* mode looks the table up through the selected
xref:metadata-types/spark-catalog.adoc[lakehouse catalog]. The Hop engine
supports the `hadoop` and `rest` catalog types. A catalog name in front of the
identifier (`lake.db.orders`) is dropped, so the same identifier works on both
engines. With a REST catalog the files are read through Hop VFS, unless the
catalog's extra configuration names another `io-impl`, for example
`org.apache.iceberg.aws.s3.S3FileIO` for credentials vended by the catalog.
+
+*Reading*
+
+* Iceberg plans the read: it skips the files and row groups that can't match,
using the statistics in the table metadata.
+* Delete files written by other engines (position and equality deletes, for
example from Flink or Spark) are applied.
+* Without a field list all columns are returned, with these Hop types:
booleans as Boolean, integers as Integer, floating point numbers as Number,
decimals as BigNumber, dates as Date, timestamps as Timestamp, binary values as
Binary and everything else as String. Timestamps without a time zone are read
as UTC. With a field list the columns are returned in the order of the list and
converted to the type of each field.
Review Comment:
**[bug]** Timestamps without a time zone are not read as UTC.
`IcebergTypeMapper.toHopValue` turns a timestamp-without-zone `LocalDateTime`
into a Hop timestamp with `Timestamp.valueOf`, which interprets that local
date-time in the JVM default zone (the same zone Parquet Input uses). Only
timestamp-with-zone values become instants. Telling readers the value is UTC
will shift data that already round-trips as wall-clock time.
**Suggestion:** Say timestamps without a zone are read as local date-times
in the Hop JVM's time zone, and that timestamps with a zone are instants. The
later sentence about time-travel timestamps using the Hop server zone can stay.
##########
docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-lake-table-output.adoc:
##########
@@ -61,13 +64,59 @@ This transform is **native Spark only**. Connectors ship
with the native Spark p
|Optional comma-separated partition column list.
|Coalesce
-|Optional partition count before write (e.g. `1` for a single file layout —
costs a shuffle).
+|Optional partition count before write (e.g. `1` for a single file layout —
costs a shuffle). Spark engine only.
|Extra options
-|Optional Spark write options as `key=value` lines (advanced).
+|Optional Spark write options as `key=value` lines (advanced). Spark engine
only.
+|===
+
+[[hop-engine]]
+== On the Hop engine
+
+On the local and remote Hop engines the transform writes Apache Iceberg
tables. Delta Lake tables give an error that asks you to use the Spark engine
for now.
+
+*One commit per pipeline run*
+
+Every copy of the transform writes its own Parquet files. Nothing is visible
in the table until the whole pipeline has finished:
+
+* When the pipeline ends without errors, all files of the run are committed as
one Iceberg snapshot. The snapshot summary records the pipeline name
(`hop.pipeline`) and the run's log channel id (`hop.run-id`).
+* When the pipeline fails or is stopped, the files are deleted and the table
is left as it was. This includes a failure in a transform *after* Lake table
output: the run is rolled back, like a database transaction.
+
+*Save modes*
+
+[options="header",cols="1,2,2"]
|===
+|Save mode |Table exists |Table doesn't exist
+
+|ErrorIfExists
+|The pipeline fails, the table is left alone
+|The table is created
+
+|Append
+|The rows are added
+|The table is created
+
+|Overwrite
+|All rows of the table are replaced by the rows of this run. The run fails
instead if another writer changed the table since the run started.
+|The table is created
+
+|Ignore
+|Nothing is written; the rows are still passed on
+|The table is created
+|===
+
+A new table gets its columns from the incoming rows (Integer as `long`, Number
as `double`, BigNumber as `decimal` with the field's length and precision,
String as `string`, Boolean as `boolean`, Date and Timestamp as `timestamp`
without time zone, in UTC, Binary as `binary`) and identity partitions on the
*Partition by* columns. It is created in the same commit as its first rows, so
a failed run doesn't leave an empty table behind. When the table exists, the
rows are matched to its columns by name.
Review Comment:
**[bug]** Date and Timestamp are not written in UTC. `toIcebergType` maps
both to `timestamp` without zone, and `toIcebergValue` stores
`LocalDateTime.ofInstant(instant, ZoneId.systemDefault())`. The clock time in
the table is the pipeline JVM's zone, not UTC. A reader in another zone, or a
user who converts to UTC before writing, will shift the value.
**Suggestion:** Say Date and Timestamp are stored as timestamp without time
zone, using the wall-clock time in the Hop JVM's default zone. Same sentence:
BigNumber uses the field length and precision only when length is greater than
0; otherwise the column is `decimal(38, 10)`.
##########
integration-tests/iceberg/0004-validation.hpl:
##########
@@ -0,0 +1,167 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<pipeline>
+ <info>
+ <name>0004-validation</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Validate 0001 extract against golden</description>
+ <pipeline_type>Normal</pipeline_type>
Review Comment:
**[nit]** The description still says "Validate 0001 extract against golden",
copied from the Spark test. This pipeline reads
`/tmp/hop-it-iceberg/0004-extract/`.
**Suggestion:** Change the description to the 0004 extract so a failed run
is not traced to the wrong workflow.
##########
docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-lake-table-output.adoc:
##########
@@ -61,13 +64,59 @@ This transform is **native Spark only**. Connectors ship
with the native Spark p
|Optional comma-separated partition column list.
|Coalesce
-|Optional partition count before write (e.g. `1` for a single file layout —
costs a shuffle).
+|Optional partition count before write (e.g. `1` for a single file layout —
costs a shuffle). Spark engine only.
|Extra options
-|Optional Spark write options as `key=value` lines (advanced).
+|Optional Spark write options as `key=value` lines (advanced). Spark engine
only.
+|===
+
+[[hop-engine]]
+== On the Hop engine
+
+On the local and remote Hop engines the transform writes Apache Iceberg
tables. Delta Lake tables give an error that asks you to use the Spark engine
for now.
+
+*One commit per pipeline run*
+
+Every copy of the transform writes its own Parquet files. Nothing is visible
in the table until the whole pipeline has finished:
+
+* When the pipeline ends without errors, all files of the run are committed as
one Iceberg snapshot. The snapshot summary records the pipeline name
(`hop.pipeline`) and the run's log channel id (`hop.run-id`).
+* When the pipeline fails or is stopped, the files are deleted and the table
is left as it was. This includes a failure in a transform *after* Lake table
output: the run is rolled back, like a database transaction.
+
+*Save modes*
+
+[options="header",cols="1,2,2"]
|===
+|Save mode |Table exists |Table doesn't exist
+
+|ErrorIfExists
+|The pipeline fails, the table is left alone
+|The table is created
+
+|Append
+|The rows are added
+|The table is created
+
+|Overwrite
+|All rows of the table are replaced by the rows of this run. The run fails
instead if another writer changed the table since the run started.
+|The table is created
+
+|Ignore
+|Nothing is written; the rows are still passed on
+|The table is created
+|===
+
+A new table gets its columns from the incoming rows (Integer as `long`, Number
as `double`, BigNumber as `decimal` with the field's length and precision,
String as `string`, Boolean as `boolean`, Date and Timestamp as `timestamp`
without time zone, in UTC, Binary as `binary`) and identity partitions on the
*Partition by* columns. It is created in the same commit as its first rows, so
a failed run doesn't leave an empty table behind. When the table exists, the
rows are matched to its columns by name.
+
+Unlike the Spark engine, Overwrite on the Hop engine keeps the table's schema
and partitioning and only replaces its rows.
+
+*Where the table is*
+
+* *PATH* mode writes the table in the given Hop VFS folder, in the layout of
Hadoop catalogs and of the Spark engine's PATH mode
(`metadata/v<N>.metadata.json` plus `version-hint.text`). The Spark and Hop
engines can read and write the same table. A folder that holds a table managed
by a catalog is refused: write that table through its catalog in TABLE mode.
+* *TABLE* mode writes through the selected lakehouse catalog. The Hop engine
supports the `hadoop` and `rest` catalog types.
+
+On a local file system that supports hard links (the usual case on Linux,
macOS and Windows NTFS), a PATH table commit publishes the new metadata file
with an operation that fails if another writer already committed that version,
so several pipelines can write the same table: one of two conflicting commits
fails and is retried. On other file systems, including HDFS, object stores like
S3 and local file systems without hard links, a PATH table should have one
writer at a time. When several pipelines write the same table at the same time
there, use a REST catalog.
Review Comment:
**[suggestion]** Local file systems without hard links are still safe for
concurrent PATH commits. `PathTableOperations.publishLocal` falls back from
`Files.createLink` to `CREATE_NEW` (`O_EXCL`), which that class treats as
atomic. Only non-local VFS (`publishRemote`: check, then write) is not atomic
across processes. That is HDFS, S3, and other Hop VFS locations, not a local
disk that simply has no hard links.
**Suggestion:** Say any `file://` location allows concurrent writers (hard
link, or exclusive create where links are unsupported). Ask for one writer, or
a REST catalog, only for HDFS, object stores, and other non-local VFS locations.
##########
docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-lake-table-input.adoc:
##########
@@ -16,19 +16,24 @@ under the License.
////
:documentationPath: /pipeline/transforms/
:language: en_US
-:description: Spark lake table input reads Delta Lake or Apache Iceberg tables
on the native Spark pipeline engine, with optional time travel and projection.
-:page-engines: Hop Engine=no, Single Threaded=no, Native Spark=native, Beam
Spark=no, Beam Flink=no, Beam Dataflow=no
-:page-marketplace: hop-engines-spark
+:description: Lake table input reads Apache Iceberg tables on the Hop engine
and Delta Lake or Apache Iceberg tables on the native Spark engine, with
optional time travel and projection.
+:page-engines: Hop Engine=yes, Single Threaded=no, Native Spark=native, Beam
Spark=no, Beam Flink=no, Beam Dataflow=no
+:page-marketplace: hop-tech-lakehouse
Review Comment:
**[suggestion]** `:page-marketplace: hop-tech-lakehouse` is not an
installable marketplace id. `optional-plugins.yaml` lists `hop-engines-spark`
and does not list `hop-tech-lakehouse`, and that plugin zip is only added to
the client by the non-release `full` profile. The same attribute is set on the
output page and on `spark-catalog.adoc`. The previous value was a catalogued
plugin; this one is not.
**Suggestion:** Add `hop-tech-lakehouse` to `optional-plugins.yaml` (with
install path) before pointing these pages at it, or keep a marketplace id that
`hop marketplace` can resolve and say in the page which engine that plugin
covers.
--
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]
