mattcasters commented on code in PR #8795:
URL: https://github.com/apache/hop/pull/8795#discussion_r4233072979
##########
plugins/engines/spark/src/main/java/org/apache/hop/spark/table/SparkLakeTableSupport.java:
##########
@@ -107,15 +110,140 @@ public static String toTableLocationUri(String path) {
}
/**
- * Spark SQL multi-part identifier for an Iceberg path-based table under the
hop path catalog:
- * {@code hop_iceberg.`file:///path/to/table`}.
+ * An Iceberg table at a path, seen through the Hadoop catalog whose
warehouse is the table's
+ * parent folder. A Hadoop catalog keeps a table of the default namespace in
{@code
+ * <warehouse>/<table>}, so the table is read and written exactly at its
path.
+ *
+ * <p>A quoted path in a single shared catalog, like {@code
hop_iceberg.`file:///data/orders`},
+ * doesn't work: Iceberg only treats an identifier as a location when Spark
creates it from {@code
+ * DataFrameReader.load(path)}. Otherwise the Hadoop catalog takes the whole
URI as a table name
+ * and stores the table under its own warehouse.
+ */
+ public record IcebergPathTable(String catalogName, String warehouse, String
tableName) {
+
+ /** Spark SQL identifier of the table, e.g. {@code
hop_iceberg_1a2b3c4d5e6f.`orders`}. */
+ public String sqlIdentifier() {
+ return catalogName + "." + procedureTableRef();
+ }
+
+ /** The table as named in an Iceberg procedure call of {@link
#catalogName()}. */
+ public String procedureTableRef() {
+ return "`" + tableName.replace("`", "``") + "`";
+ }
+ }
+
+ /**
+ * The path catalog and table name for an Iceberg table at {@code path}.
Tables in the same folder
+ * share a catalog.
+ */
+ public static IcebergPathTable icebergPathTable(String path) {
+ String uri = canonicalLocation(toTableLocationUri(path));
+ int slash = uri == null ? -1 : uri.lastIndexOf('/');
+ String parent = slash < 0 ? "" : uri.substring(0, slash);
+ String name = slash < 0 ? "" : uri.substring(slash + 1);
+ if (name.isEmpty() || isSchemeOnly(parent)) {
+ throw new IllegalArgumentException(
+ "Iceberg table path '" + path + "' needs a parent folder, e.g.
file:///data/orders");
+ }
+ return new IcebergPathTable(
+ SparkLakeFormats.ICEBERG_PATH_CATALOG_NAME + "_" + shortHash(parent),
parent, name);
+ }
+
+ /**
+ * Like {@link #icebergPathTable(String)}, but reports an invalid path as a
{@link HopException}
+ * that names the transform.
+ */
+ public static IcebergPathTable icebergPathTable(String path, String
transformName)
+ throws HopException {
+ try {
+ return icebergPathTable(path);
+ } catch (IllegalArgumentException e) {
+ throw new HopException("Spark Lake Table '" + transformName + "': " +
e.getMessage(), e);
+ }
+ }
+
+ private static final Pattern URI_SCHEME =
Pattern.compile("^([A-Za-z][A-Za-z0-9+.-]*):(.*)$");
+
+ /**
+ * One spelling per location, so that one folder is always one catalog: the
scheme is lowercased,
+ * {@code file:/x}, {@code file://localhost/x} and {@code file:///x} all
become {@code file:///x},
+ * empty path segments are collapsed, and {@code .} and {@code ..} are
resolved. A trailing slash
+ * is removed.
+ */
+ static String canonicalLocation(String uri) {
+ if (StringUtils.isEmpty(uri)) {
+ return uri;
+ }
+ Matcher matcher = URI_SCHEME.matcher(uri.trim());
+ if (!matcher.matches() || matcher.group(1).length() == 1) {
+ // No scheme (or a Windows drive letter): leave it as it is.
+ return StringUtils.removeEnd(uri.trim(), "/");
+ }
+ String scheme = matcher.group(1).toLowerCase(java.util.Locale.ROOT);
+ String rest = matcher.group(2);
+ String authority = "";
+ String pathPart = rest;
+ if (rest.startsWith("//")) {
+ int end = rest.indexOf('/', 2);
+ authority = end < 0 ? rest.substring(2) : rest.substring(2, end);
+ pathPart = end < 0 ? "" : rest.substring(end);
+ }
+ if ("file".equals(scheme) && "localhost".equalsIgnoreCase(authority)) {
+ authority = "";
+ }
+ java.util.Deque<String> segments = new java.util.ArrayDeque<>();
+ for (String segment : pathPart.split("/")) {
+ if (segment.isEmpty() || ".".equals(segment)) {
+ continue;
+ }
+ if ("..".equals(segment)) {
+ segments.pollLast();
+ } else {
+ segments.addLast(segment);
Review Comment:
**[suggestion]** A Windows drive letter is still two folders. `Path.toUri()`
produces `file:///C:/data/orders`, while a pasted `file:///c:/data/orders`
keeps the lowercase `c:` segment, so the same directory gets two
`hop_iceberg_<hash>` catalogs. When the scheme is `file`, uppercase a
single-letter drive segment (`c:` → `C:`) before it is hashed.
##########
plugins/engines/spark/src/main/java/org/apache/hop/spark/table/SparkLakeTableSupport.java:
##########
@@ -107,15 +110,140 @@ public static String toTableLocationUri(String path) {
}
/**
- * Spark SQL multi-part identifier for an Iceberg path-based table under the
hop path catalog:
- * {@code hop_iceberg.`file:///path/to/table`}.
+ * An Iceberg table at a path, seen through the Hadoop catalog whose
warehouse is the table's
+ * parent folder. A Hadoop catalog keeps a table of the default namespace in
{@code
+ * <warehouse>/<table>}, so the table is read and written exactly at its
path.
+ *
+ * <p>A quoted path in a single shared catalog, like {@code
hop_iceberg.`file:///data/orders`},
+ * doesn't work: Iceberg only treats an identifier as a location when Spark
creates it from {@code
+ * DataFrameReader.load(path)}. Otherwise the Hadoop catalog takes the whole
URI as a table name
+ * and stores the table under its own warehouse.
+ */
+ public record IcebergPathTable(String catalogName, String warehouse, String
tableName) {
+
+ /** Spark SQL identifier of the table, e.g. {@code
hop_iceberg_1a2b3c4d5e6f.`orders`}. */
+ public String sqlIdentifier() {
+ return catalogName + "." + procedureTableRef();
+ }
+
+ /** The table as named in an Iceberg procedure call of {@link
#catalogName()}. */
+ public String procedureTableRef() {
+ return "`" + tableName.replace("`", "``") + "`";
+ }
+ }
+
+ /**
+ * The path catalog and table name for an Iceberg table at {@code path}.
Tables in the same folder
+ * share a catalog.
+ */
+ public static IcebergPathTable icebergPathTable(String path) {
+ String uri = canonicalLocation(toTableLocationUri(path));
+ int slash = uri == null ? -1 : uri.lastIndexOf('/');
+ String parent = slash < 0 ? "" : uri.substring(0, slash);
+ String name = slash < 0 ? "" : uri.substring(slash + 1);
+ if (name.isEmpty() || isSchemeOnly(parent)) {
+ throw new IllegalArgumentException(
+ "Iceberg table path '" + path + "' needs a parent folder, e.g.
file:///data/orders");
+ }
+ return new IcebergPathTable(
+ SparkLakeFormats.ICEBERG_PATH_CATALOG_NAME + "_" + shortHash(parent),
parent, name);
+ }
+
+ /**
+ * Like {@link #icebergPathTable(String)}, but reports an invalid path as a
{@link HopException}
+ * that names the transform.
+ */
+ public static IcebergPathTable icebergPathTable(String path, String
transformName)
+ throws HopException {
+ try {
+ return icebergPathTable(path);
+ } catch (IllegalArgumentException e) {
+ throw new HopException("Spark Lake Table '" + transformName + "': " +
e.getMessage(), e);
+ }
+ }
+
+ private static final Pattern URI_SCHEME =
Pattern.compile("^([A-Za-z][A-Za-z0-9+.-]*):(.*)$");
+
+ /**
+ * One spelling per location, so that one folder is always one catalog: the
scheme is lowercased,
+ * {@code file:/x}, {@code file://localhost/x} and {@code file:///x} all
become {@code file:///x},
+ * empty path segments are collapsed, and {@code .} and {@code ..} are
resolved. A trailing slash
+ * is removed.
+ */
+ static String canonicalLocation(String uri) {
+ if (StringUtils.isEmpty(uri)) {
+ return uri;
+ }
+ Matcher matcher = URI_SCHEME.matcher(uri.trim());
+ if (!matcher.matches() || matcher.group(1).length() == 1) {
+ // No scheme (or a Windows drive letter): leave it as it is.
+ return StringUtils.removeEnd(uri.trim(), "/");
+ }
+ String scheme = matcher.group(1).toLowerCase(java.util.Locale.ROOT);
+ String rest = matcher.group(2);
+ String authority = "";
+ String pathPart = rest;
+ if (rest.startsWith("//")) {
+ int end = rest.indexOf('/', 2);
+ authority = end < 0 ? rest.substring(2) : rest.substring(2, end);
+ pathPart = end < 0 ? "" : rest.substring(end);
+ }
+ if ("file".equals(scheme) && "localhost".equalsIgnoreCase(authority)) {
+ authority = "";
+ }
+ java.util.Deque<String> segments = new java.util.ArrayDeque<>();
+ for (String segment : pathPart.split("/")) {
+ if (segment.isEmpty() || ".".equals(segment)) {
+ continue;
+ }
+ if ("..".equals(segment)) {
+ segments.pollLast();
+ } else {
+ segments.addLast(segment);
+ }
+ }
+ String normalizedPath = segments.isEmpty() ? "" : "/" + String.join("/",
segments);
+ return scheme + "://" + authority + normalizedPath;
Review Comment:
**[bug]** `canonicalLocation` is the string `icebergPathTable` uses as the
Hadoop warehouse and the table name, and it never percent-decodes. Scheme-less
paths go through `toTableLocationUri` → `Path.toUri()`, so `/data/my
lake/orders` becomes `file:///data/my%20lake/orders` (a table folder `my
orders` becomes `my%20orders`). Hadoop's `Path` does not treat `%` as an
escape; `new Path("file:///data/my%20lake")` is the directory `/data/my%20lake`
(`file:///data/my%2520lake`), not `/data/my lake`. The same split of
`file:///data/my lake/orders` and `file:///data/my%20lake/orders` also hashes
to two catalogs. `SparkLakeTableIcebergPathTest` uses a temp path with no such
characters, so it stays green.
**Suggestion:** Percent-decode each path segment and the authority before
hashing and before setting `spark.sql.catalog.<name>.warehouse`. Decode with
`URI.getPath()` (not `URLDecoder`, so `+` stays `+`); leave a segment as-is
when it has a raw space or a broken `%` escape. `/data/my lake/my orders`,
`file:///data/my%20lake/my%20orders`, and `file:///data/my lake/my orders`
should all yield warehouse `file:///data/my lake` and table name `my orders`.
Extend `oneFolderIsOneCatalogWhateverTheSpelling` with that case.
--
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]
