This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 056876feb06 AddFiles: PathPattern and RouteFiles (#40749)
056876feb06 is described below
commit 056876feb06f08489b95aa7c450daf0bc109d5d4
Author: claudevdm <[email protected]>
AuthorDate: Mon Oct 26 21:18:18 2026 -0400
AddFiles: PathPattern and RouteFiles (#40749)
* Pattern match
* Route files
* comments
---
.../apache/beam/sdk/io/iceberg/PathPattern.java | 247 +++++++++++++++++
.../org/apache/beam/sdk/io/iceberg/RouteFiles.java | 70 +++++
.../beam/sdk/io/iceberg/PathPatternTest.java | 306 +++++++++++++++++++++
.../apache/beam/sdk/io/iceberg/RouteFilesTest.java | 125 +++++++++
4 files changed, 748 insertions(+)
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java
new file mode 100644
index 00000000000..33ed4c60852
--- /dev/null
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PathPattern.java
@@ -0,0 +1,247 @@
+/*
+ * 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.beam.sdk.io.iceberg;
+
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
+import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.beam.sdk.io.FileSystemUtils;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Matches a file path against a glob with {@code {name}} placeholders, and
reports what each
+ * placeholder captured.
+ *
+ * <p>Pattern {@code gs://lake/export/{table}/**} over path {@code
+ * gs://lake/export/orders/part-0.parquet} captures {@code table=orders}. The
whole path has to
+ * match, so the pattern decides both which table a file belongs to and
whether it belongs at all.
+ *
+ * <p>A pattern may hold at most one {@code **}, before or after its
placeholders. A second one can
+ * leave a placeholder between the two with nothing fixing which directory it
names, so two are
+ * refused. {@code *} and {@code ?} are unrestricted: neither crosses a
directory, so neither can
+ * move a placeholder.
+ *
+ * <p>Two edges follow from Beam's glob translation. A {@code **} matches one
directory or more,
+ * never none, so a pattern with a {@code **} above its placeholder does not
match a file with no
+ * directory in that position. And a placeholder with nothing after it
captures the file name rather
+ * than a directory.
+ */
+final class PathPattern implements Serializable {
+ private static final Pattern PLACEHOLDER = Pattern.compile("\\{(.+?)}");
+
+ /** A name must survive rendering into a dotted table template: no dots,
slashes or blanks. */
+ private static final Pattern NAME = Pattern.compile("[A-Za-z0-9_-]+");
+
+ private final Pattern regex;
+ private final List<String> captureNames;
+
+ private PathPattern(Pattern regex, List<String> captureNames) {
+ this.regex = regex;
+ this.captureNames = Collections.unmodifiableList(captureNames);
+ }
+
+ /**
+ * Translates the path pattern to a regex, each placeholder becoming a
capturing group. {@code
+ * gs://lake/{db}/{tbl}/**} becomes {@code gs://lake/([^/]+)/([^/]+)/.*},
which a path has to
+ * match in full.
+ */
+ static PathPattern compile(String pathPattern) {
+ List<String> names = placeholders(pathPattern);
+ checkNamesAreBindable(names, pathPattern);
+ // the -1 keeps a trailing empty part, so "gs://lake/{tbl}" still yields a
group
+ String[] fixedParts = PLACEHOLDER.split(pathPattern, -1);
+ checkAtMostOneDoubleStar(pathPattern);
+ checkArgument(
+ !pathPattern.endsWith("/"),
+ "'%s' ends with '/', so it names a folder and matches no file; end it
with a selector for"
+ + " the files, such as '**'.",
+ pathPattern);
+ Pattern regex = Pattern.compile(toRegex(fixedParts));
+ checkState(
+ regex.matcher("").groupCount() == names.size(),
+ "'%s' translated to a regex with %s groups for %s placeholders.",
+ pathPattern,
+ regex.matcher("").groupCount(),
+ names.size());
+ return new PathPattern(regex, names);
+ }
+
+ private static void checkNamesAreBindable(List<String> names, String
pathPattern) {
+ for (int i = 0; i < names.size(); i++) {
+ String name = names.get(i);
+ checkArgument(
+ NAME.matcher(name).matches(),
+ "Placeholder '%s' in '%s' is not a name; use letters, digits, '_' or
'-'.",
+ name,
+ pathPattern);
+ checkArgument(
+ names.indexOf(name) == i,
+ "Placeholder '%s' appears twice in '%s'; a name can only be captured
once.",
+ name,
+ pathPattern);
+ }
+ }
+
+ // Two '**' can sit either side of a placeholder, and then nothing fixes
which directory it
+ // names: gs://lake/**/{tbl}/** binds tbl to a different folder for every
path depth. One '**'
+ // cannot do that wherever it sits, since it is only ever on one side of a
given placeholder,
+ // so one is the limit. A '*' or '?' never crosses a directory and is not
restricted.
+ private static void checkAtMostOneDoubleStar(String pathPattern) {
+ int first = pathPattern.indexOf("**");
+ checkArgument(
+ first < 0 || pathPattern.indexOf("**", first + 2) < 0,
+ "'%s' holds more than one '**'. One is allowed, before or after the
placeholders; a"
+ + " second can leave a placeholder between them with no directory
fixing it. Use '*'"
+ + " for a single directory level.",
+ pathPattern);
+ }
+
+ /**
+ * Splits the pattern at its placeholders, expands the fixed text in between
as a glob, and joins
+ * those parts back with a capturing group wherever a placeholder was.
+ *
+ * <p>Globbing is Beam's own, through {@link FileSystemUtils}: {@code *}
stays inside one
+ * directory level, {@code **} crosses them, {@code ?} is one character.
None of them capture, so
+ * group numbers line up with {@code captureNames}.
+ */
+ private static String toRegex(String[] fixedParts) {
+ List<String> expanded = new ArrayList<>();
+ for (String fixedText : fixedParts) {
+ expanded.add(FileSystemUtils.wildcardToRegexp(fixedText));
+ }
+ return String.join("([^/]+)", expanded);
+ }
+
+ List<String> captureNames() {
+ return captureNames;
+ }
+
+ /** Captures by name plus the path below them, or null when the path does
not match. */
+ @Nullable Match match(String path) {
+ Matcher matcher = regex.matcher(path);
+ if (!matcher.matches()) {
+ return null;
+ }
+ Map<String, String> captures = new LinkedHashMap<>();
+ for (int i = 0; i < captureNames.size(); i++) {
+ captures.put(captureNames.get(i), checkStateNotNull(matcher.group(i +
1)));
+ }
+ return new Match(captures, remainderBelow(matcher, path));
+ }
+
+ /**
+ * The path below the last capture, file name included, so a {@code
{db}/{tbl}/**} pattern over
+ * {@code gs://lake/sales/orders/y=2024/part-0.parquet} leaves {@code
y=2024/part-0.parquet}.
+ *
+ * <p>Empty when the last capture is not followed by a separator, which is
the {@code
+ * {table}.parquet} layout: there are no folders below the table, so there
is nothing to inspect.
+ */
+ private String remainderBelow(Matcher matcher, String path) {
+ if (captureNames.isEmpty()) {
+ return "";
+ }
+ int end = matcher.end(captureNames.size());
+ if (end >= path.length() || path.charAt(end) != '/') {
+ return "";
+ }
+ return path.substring(end + 1);
+ }
+
+ /** What one path matched. */
+ static final class Match {
+ private final Map<String, String> captures;
+ private final String remainder;
+
+ private Match(Map<String, String> captures, String remainder) {
+ this.captures = Collections.unmodifiableMap(captures);
+ this.remainder = remainder;
+ }
+
+ /** In the order the placeholders appear, so a message about them reads
like the pattern. */
+ Map<String, String> captures() {
+ return captures;
+ }
+
+ String remainder() {
+ return remainder;
+ }
+ }
+
+ /**
+ * Fills a table template from a path's captures, so {@code
warehouse.{db}.{tbl}} with {@code
+ * db=sales, tbl=orders} gives {@code warehouse.sales.orders}.
+ */
+ static String render(String template, Map<String, String> captures) {
+ // split with -1 gives one more literal part than there are placeholders,
so literals[i + 1]
+ // is always the text following names.get(i)
+ String[] literals = PLACEHOLDER.split(template, -1);
+ List<String> names = placeholders(template);
+ StringBuilder rendered = new StringBuilder(literals[0]);
+ for (int i = 0; i < names.size(); i++) {
+ String name = names.get(i);
+ String value =
+ checkArgumentNotNull(
+ captures.get(name),
+ "Table '%s' uses {%s}, which the path did not capture; captured
%s.",
+ template,
+ name,
+ captures.keySet());
+ rendered.append(value).append(literals[i + 1]);
+ }
+ return rendered.toString();
+ }
+
+ /**
+ * Refuses a template this pattern cannot fill. Meant to run once at
construction, so that a
+ * {@code table} naming a placeholder the pattern never captures is a config
error instead of one
+ * error row per file.
+ */
+ void checkRenderable(String template) {
+ for (String name : placeholders(template)) {
+ checkArgument(
+ captureNames.contains(name),
+ "Table '%s' uses {%s}, which the path pattern does not capture; it
captures %s.",
+ template,
+ name,
+ captureNames);
+ }
+ }
+
+ /**
+ * Placeholder names in the order they appear, so {@code sales.{db}.{tbl}}
gives {@code [db,
+ * tbl]}.
+ */
+ static List<String> placeholders(String text) {
+ List<String> names = new ArrayList<>();
+ Matcher matcher = PLACEHOLDER.matcher(text);
+ while (matcher.find()) {
+ names.add(checkStateNotNull(matcher.group(1)));
+ }
+ return names;
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java
new file mode 100644
index 00000000000..5a6d7c04722
--- /dev/null
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RouteFiles.java
@@ -0,0 +1,70 @@
+/*
+ * 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.beam.sdk.io.iceberg;
+
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TupleTag;
+
+/**
+ * Names the destination table for each file path, from the pattern that
describes the layout and
+ * the template that spells the identifier.
+ *
+ * <p>A path the pattern does not match is not something this transform was
pointed at, so it
+ * becomes an error row rather than being dropped: a mistyped pattern has to
be visible, not
+ * silently register nothing.
+ */
+final class RouteFiles extends DoFn<String, KV<String, String>> {
+
+ static final TupleTag<KV<String, String>> ROUTED = new TupleTag<>();
+ static final TupleTag<Row> ERRORS = new TupleTag<>();
+
+ /** The same rows as the transform's other errors, so one output can carry
both. */
+ static final Schema ERROR_SCHEMA = AddFiles.ERROR_SCHEMA;
+
+ private final String pathPattern;
+ private final String tableTemplate;
+ private final PathPattern pattern;
+
+ RouteFiles(String pathPattern, String tableTemplate) {
+ this.pathPattern = pathPattern;
+ this.tableTemplate = tableTemplate;
+ // both throw here, at pipeline construction, rather than once per file in
a worker
+ this.pattern = PathPattern.compile(pathPattern);
+ this.pattern.checkRenderable(tableTemplate);
+ }
+
+ @ProcessElement
+ public void process(@Element String path, MultiOutputReceiver output) {
+ PathPattern.Match match = pattern.match(path);
+
+ if (match == null) {
+ String error = "did not match path_pattern '" + pathPattern + "'";
+ output.get(ERRORS).output(errorRow(path, error));
+ return;
+ }
+ String table = PathPattern.render(tableTemplate, match.captures());
+ output.get(ROUTED).output(KV.of(table, path));
+ }
+
+ private static Row errorRow(String file, String error) {
+ return Row.withSchema(ERROR_SCHEMA).addValues(file, error).build();
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PathPatternTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PathPatternTest.java
new file mode 100644
index 00000000000..c16722f01ae
--- /dev/null
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PathPatternTest.java
@@ -0,0 +1,306 @@
+/*
+ * 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.beam.sdk.io.iceberg;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class PathPatternTest {
+
+ @Test
+ public void severalPlaceholdersInPathOrder() {
+ assertEquals(
+ Arrays.asList("db", "tbl", "y"),
PathPattern.placeholders("gs://lake/{db}/{tbl}/y={y}/**"));
+ }
+
+ /** Lazily, so the first {@code }} wins; greedy would give one name {@code
db}.{tbl}. */
+ @Test
+ public void adjacentPlaceholdersStaySeparate() {
+ assertEquals(Arrays.asList("db", "tbl"),
PathPattern.placeholders("{db}.{tbl}"));
+ }
+
+ @Test
+ public void noPlaceholders() {
+ assertEquals(Collections.emptyList(),
PathPattern.placeholders("gs://lake/export/orders/"));
+ assertEquals(Collections.emptyList(), PathPattern.placeholders(""));
+ }
+
+ /** {@code .+?} needs a character, so these are literal text rather than a
malformed name. */
+ @Test
+ public void emptyOrUnterminatedBracesAreNotPlaceholders() {
+ assertEquals(Collections.emptyList(), PathPattern.placeholders("{}"));
+ assertEquals(Collections.emptyList(),
PathPattern.placeholders("gs://lake/{table"));
+ }
+
+ @Test
+ public void strayClosingBraceIsSkipped() {
+ assertEquals(Arrays.asList("a"), PathPattern.placeholders("table}/{a}"));
+ }
+
+ /** Kept, not collapsed, so compiling the pattern can refuse the duplicate.
*/
+ @Test
+ public void repeatedNameIsReportedTwice() {
+ assertEquals(Arrays.asList("t", "t"), PathPattern.placeholders("{t}/{t}"));
+ }
+
+ /**
+ * Nothing here validates a name: {@code .} spans {@code /}, {@code .} and
{@code {}, and there is
+ * no escape syntax. Compiling the pattern is what refuses these.
+ */
+ @Test
+ public void namesAreNotValidatedHere() {
+ assertEquals(Arrays.asList("a/b"), PathPattern.placeholders("{a/b}"));
+ assertEquals(Arrays.asList("a.b"), PathPattern.placeholders("{a.b}"));
+ assertEquals(Arrays.asList("a{b"), PathPattern.placeholders("{a{b}"));
+ assertEquals(Arrays.asList(" "), PathPattern.placeholders("{ }"));
+ }
+
+ private static PathPattern.Match matched(String pathPattern, String path) {
+ PathPattern.Match match = PathPattern.compile(pathPattern).match(path);
+ assertNotNull(pathPattern + " did not match " + path, match);
+ return match;
+ }
+
+ @Test
+ public void compileWithNoPlaceholders() {
+ PathPattern pattern = PathPattern.compile("gs://lake/orders/**");
+ assertEquals(Collections.emptyList(), pattern.captureNames());
+ PathPattern.Match match = pattern.match("gs://lake/orders/part-0.parquet");
+ assertNotNull(match);
+ assertEquals(Collections.emptyMap(), match.captures());
+ assertEquals("", match.remainder());
+ }
+
+ @Test
+ public void capturesTheFoldersAndTheRemainderBelowThem() {
+ PathPattern.Match match =
+ matched("gs://lake/{db}/{tbl}/**",
"gs://lake/sales/orders/y=2024/part-0.parquet");
+ assertEquals("sales", match.captures().get("db"));
+ assertEquals("orders", match.captures().get("tbl"));
+ assertEquals("y=2024/part-0.parquet", match.remainder());
+ }
+
+ /** The file name is part of the remainder, because _SUCCESS and .crc are
files. */
+ @Test
+ public void theRemainderIncludesTheFileName() {
+ PathPattern.Match match = matched("gs://lake/{tbl}/**",
"gs://lake/orders/_SUCCESS");
+ assertEquals("orders", match.captures().get("tbl"));
+ assertEquals("_SUCCESS", match.remainder());
+ }
+
+ /** Captures come back in the order the placeholders appear, so messages
read like the pattern. */
+ @Test
+ public void capturesAreInPatternOrder() {
+ PathPattern.Match match =
+ matched("gs://lake/{db}/{tbl}/**", "gs://lake/sales/orders/f.parquet");
+ assertEquals(Arrays.asList("db", "tbl"), new
ArrayList<>(match.captures().keySet()));
+ }
+
+ /** A deeper path binds the shallower folders rather than failing; '**'
absorbs the rest. */
+ @Test
+ public void placeholderNeverSpansADirectoryLevel() {
+ PathPattern.Match match =
+ matched("gs://lake/{db}/{tbl}/**",
"gs://lake/sales/eu/orders/f.parquet");
+ assertEquals("sales", match.captures().get("db"));
+ assertEquals("eu", match.captures().get("tbl"));
+ assertEquals("orders/f.parquet", match.remainder());
+ }
+
+ /** '*' is one level, so the depth is pinned where '**' would have absorbed
it. */
+ @Test
+ public void singleStarPinsTheDepth() {
+ PathPattern pattern = PathPattern.compile("gs://lake/export/{table}/*");
+ assertNotNull(pattern.match("gs://lake/export/orders/part-0.parquet"));
+ assertNull(pattern.match("gs://lake/export/orders/y=2024/part-0.parquet"));
+ }
+
+ /** The pattern is the only place on this transform that can say which files
belong. */
+ @Test
+ public void aSelectorExcludesOtherFiles() {
+ PathPattern pattern =
PathPattern.compile("gs://lake/export/{table}/*.parquet");
+ assertNotNull(pattern.match("gs://lake/export/orders/f.parquet"));
+ assertNull(pattern.match("gs://lake/export/orders/f.json"));
+ }
+
+ /** A literal folder keeps a sibling's files out of the table, even when
they are parquet. */
+ @Test
+ public void aLiteralFolderExcludesItsSiblings() {
+ PathPattern pattern =
PathPattern.compile("gs://lake/export/{tbl}/data/**");
+
assertNotNull(pattern.match("gs://lake/export/orders/data/part-0.parquet"));
+ assertNull(pattern.match("gs://lake/export/orders/rejected/bad.parquet"));
+ }
+
+ /** One file per table is a layout too, and it is unambiguous under a
whole-path match. */
+ @Test
+ public void theFileNameCanBeTheTable() {
+ PathPattern.Match match =
+ matched("gs://lake/export/{table}.parquet",
"gs://lake/export/orders.parquet");
+ assertEquals("orders", match.captures().get("table"));
+ assertEquals("", match.remainder());
+ }
+
+ @Test
+ public void anchoredAtTheStart() {
+
assertNull(PathPattern.compile("gs://lake/{tbl}/**").match("gs://other/orders/f.parquet"));
+ }
+
+ /** Unquoted, {@code a.b+c} would also match {@code axbbc}. */
+ @Test
+ public void literalTextIsQuoted() {
+ PathPattern pattern = PathPattern.compile("gs://lake/a.b+c/{tbl}/**");
+ assertNotNull(pattern.match("gs://lake/a.b+c/orders/f.parquet"));
+ assertNull(pattern.match("gs://lake/axbbc/orders/f.parquet"));
+ }
+
+ /** Globs must not capture, or the names would bind to the wrong folders. */
+ @Test
+ public void globsDoNotBindToPlaceholders() {
+ PathPattern.Match match =
+ matched("gs://lake/{db}/*/v?/{tbl}/**",
"gs://lake/sales/skip/v2/orders/f.parquet");
+ assertEquals("sales", match.captures().get("db"));
+ assertEquals("orders", match.captures().get("tbl"));
+ assertEquals(2, match.captures().size());
+ }
+
+ /** Ending at a folder would match a directory marker and never a file. */
+ @Test
+ public void patternEndingWithASlashIsRefused() {
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{tbl}/"));
+ }
+
+ /** A '**' above a placeholder would swallow the folder the placeholder is
meant to name. */
+ @Test
+ public void moreThanOneDoubleStarIsRefused() {
+ assertThrows(
+ IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/**/{tbl}/**"));
+ assertThrows(
+ IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{db}/**/{tbl}/**"));
+ assertThrows(
+ IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/**/**/{tbl}/*"));
+ }
+
+ @Test
+ public void oneDoubleStarMayComeBeforeThePlaceholder() {
+ PathPattern.Match match =
+ matched("gs://lake/**/{tbl}/*.parquet",
"gs://lake/a/b/orders/part-0.parquet");
+ assertEquals("orders", match.captures().get("tbl"));
+ }
+
+ /** A '**' stands for one directory or more, so the table cannot sit at the
pattern's root. */
+ @Test
+ public void aDoubleStarDoesNotMatchZeroDirectories() {
+ PathPattern pattern = PathPattern.compile("gs://lake/**/{tbl}/*.parquet");
+ assertNull(pattern.match("gs://lake/orders/part-0.parquet"));
+ }
+
+ /** Beam's glob translator leaves brackets unescaped, so they act as a
character class. */
+ @Test
+ public void bracketsInLiteralTextAreACharacterClass() {
+ PathPattern pattern = PathPattern.compile("gs://lake/a[bc]/{tbl}/**");
+ assertNotNull(pattern.match("gs://lake/ab/orders/part-0.parquet"));
+ assertNull(pattern.match("gs://lake/a[bc]/orders/part-0.parquet"));
+ }
+
+ @Test
+ public void namesThatCannotBeBoundAreRefused() {
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{a/b}/**"));
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{a.b}/**"));
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{a{b}/**"));
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{ }/**"));
+ }
+
+ @Test
+ public void repeatedNameIsRefused() {
+ assertThrows(IllegalArgumentException.class, () ->
PathPattern.compile("gs://lake/{t}/{t}/**"));
+ }
+
+ /** What A1 reports as literal text stays literal once compiled. */
+ @Test
+ public void emptyOrUnterminatedBracesCompileAsLiteralText() {
+ PathPattern empty = PathPattern.compile("gs://lake/{}/**");
+ assertEquals(Collections.emptyList(), empty.captureNames());
+ assertNotNull(empty.match("gs://lake/{}/f.parquet"));
+ assertEquals(Collections.emptyList(),
PathPattern.compile("gs://lake/{tbl/**").captureNames());
+ }
+
+ @Test
+ public void renderFillsTheTemplateFromTheCaptures() {
+ PathPattern.Match match =
+ matched("gs://lake/{db}/{tbl}/**", "gs://lake/sales/orders/f.parquet");
+ assertEquals(
+ "warehouse.sales.orders", PathPattern.render("warehouse.{db}.{tbl}",
match.captures()));
+ }
+
+ /** Templates at both extremes: no placeholder at all, and nothing but a
placeholder. */
+ @Test
+ public void renderHandlesTemplatesWithoutAnyLiteralOrAnyPlaceholder() {
+ PathPattern.Match match = matched("gs://lake/{tbl}/**",
"gs://lake/orders/f.parquet");
+ assertEquals("lake.events", PathPattern.render("lake.events",
match.captures()));
+ assertEquals("orders", PathPattern.render("{tbl}", match.captures()));
+ }
+
+ /** A name may repeat in the template, even though the path pattern refuses
a repeat. */
+ @Test
+ public void renderRepeatsANameAsOftenAsTheTemplateUsesIt() {
+ PathPattern.Match match = matched("gs://lake/{tbl}/**",
"gs://lake/orders/f.parquet");
+ assertEquals("orders.orders_v2", PathPattern.render("{tbl}.{tbl}_v2",
match.captures()));
+ }
+
+ /** Folder names can hold {@code $} and {@code \}, which appendReplacement
would have eaten. */
+ @Test
+ public void renderTreatsCaptureValuesAsLiteralText() {
+ PathPattern.Match match = matched("gs://lake/{tbl}/**",
"gs://lake/q$1/f.parquet");
+ assertEquals("q$1", match.captures().get("tbl"));
+ assertEquals("lake.q$1", PathPattern.render("lake.{tbl}",
match.captures()));
+ }
+
+ /** The per-element backstop; checkRenderable is meant to catch this at
construction. */
+ @Test
+ public void renderRefusesAnUnboundPlaceholder() {
+ PathPattern.Match match = matched("gs://lake/{db}/**",
"gs://lake/sales/f.parquet");
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> PathPattern.render("warehouse.{db}.{tbl}", match.captures()));
+ }
+
+ /** Every template placeholder captured, only some captured, and none -- all
legal. */
+ @Test
+ public void checkRenderableAcceptsAnyTemplateThePatternCanFill() {
+ PathPattern pattern = PathPattern.compile("gs://lake/{db}/{tbl}/**");
+ pattern.checkRenderable("warehouse.{db}.{tbl}");
+ pattern.checkRenderable("warehouse.{tbl}");
+ pattern.checkRenderable("lake.events");
+ }
+
+ @Test
+ public void checkRenderableRefusesAnUncapturedName() {
+ PathPattern pattern = PathPattern.compile("gs://lake/{db}/{tbl}/**");
+ assertThrows(
+ IllegalArgumentException.class, () ->
pattern.checkRenderable("warehouse.{db}.{table}"));
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java
new file mode 100644
index 00000000000..f10e9abc870
--- /dev/null
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RouteFilesTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.beam.sdk.io.iceberg;
+
+import static org.junit.Assert.assertThrows;
+
+import java.util.Arrays;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class RouteFilesTest {
+
+ @Rule public TestPipeline pipeline = TestPipeline.create();
+
+ private PCollectionTuple route(String pathPattern, String table, String...
paths) {
+ PCollectionTuple out =
+ pipeline
+ .apply(Create.of(Arrays.asList(paths)))
+ .apply(
+ ParDo.of(new RouteFiles(pathPattern, table))
+ .withOutputTags(RouteFiles.ROUTED,
TupleTagList.of(RouteFiles.ERRORS)));
+ out.get(RouteFiles.ERRORS).setRowSchema(RouteFiles.ERROR_SCHEMA);
+ return out;
+ }
+
+ @Test
+ public void routesEachFileToTheTableItsFoldersName() {
+ PCollectionTuple out =
+ route(
+ "gs://lake/{db}/{tbl}/**",
+ "warehouse.{db}.{tbl}",
+ "gs://lake/sales/orders/part-0.parquet",
+ "gs://lake/sales/orders/part-1.parquet",
+ "gs://lake/eu/customers/part-0.parquet");
+
+ PAssert.that(out.get(RouteFiles.ROUTED))
+ .containsInAnyOrder(
+ KV.of("warehouse.sales.orders",
"gs://lake/sales/orders/part-0.parquet"),
+ KV.of("warehouse.sales.orders",
"gs://lake/sales/orders/part-1.parquet"),
+ KV.of("warehouse.eu.customers",
"gs://lake/eu/customers/part-0.parquet"));
+ PAssert.that(out.get(RouteFiles.ERRORS)).empty();
+
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** A mistyped pattern has to be visible, so an unmatched path is a row and
not a drop. */
+ @Test
+ public void aPathThatDoesNotMatchBecomesAnErrorRow() {
+ PCollectionTuple out =
+ route(
+ "gs://lake/{db}/{tbl}/**",
+ "warehouse.{db}.{tbl}",
+ "gs://lake/sales/orders/part-0.parquet",
+ "gs://lake/stray.parquet");
+
+ PAssert.that(out.get(RouteFiles.ROUTED))
+ .containsInAnyOrder(
+ KV.of("warehouse.sales.orders",
"gs://lake/sales/orders/part-0.parquet"));
+ PAssert.that(out.get(RouteFiles.ERRORS))
+ .containsInAnyOrder(
+ Row.withSchema(RouteFiles.ERROR_SCHEMA)
+ .addValues(
+ "gs://lake/stray.parquet",
+ "did not match path_pattern 'gs://lake/{db}/{tbl}/**'")
+ .build());
+
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** A template using none of the captures is legal: every folder merges into
one table. */
+ @Test
+ public void aTemplateWithoutPlaceholdersSendsEverythingToOneTable() {
+ PCollectionTuple out =
+ route(
+ "gs://lake/{tbl}/**", "lake.events",
+ "gs://lake/a/f.parquet", "gs://lake/b/f.parquet");
+
+ PAssert.that(out.get(RouteFiles.ROUTED))
+ .containsInAnyOrder(
+ KV.of("lake.events", "gs://lake/a/f.parquet"),
+ KV.of("lake.events", "gs://lake/b/f.parquet"));
+
+ pipeline.run().waitUntilFinish();
+ }
+
+ /** Config errors belong at construction, not once per file in a worker. */
+ @Test
+ public void aMalformedPatternFailsAtConstruction() {
+ assertThrows(
+ IllegalArgumentException.class, () -> new
RouteFiles("gs://lake/{tbl}/", "lake.{tbl}"));
+ }
+
+ @Test
+ public void aTemplateNamingAnUncapturedPlaceholderFailsAtConstruction() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new RouteFiles("gs://lake/{db}/{tbl}/**",
"warehouse.{database}"));
+ }
+}
