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 000000000000..33ed4c608521 --- /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. + * + *

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. + * + *

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. + * + *

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 captureNames; + + private PathPattern(Pattern regex, List 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 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 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. + * + *

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 expanded = new ArrayList<>(); + for (String fixedText : fixedParts) { + expanded.add(FileSystemUtils.wildcardToRegexp(fixedText)); + } + return String.join("([^/]+)", expanded); + } + + List 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 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}. + * + *

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 captures; + private final String remainder; + + private Match(Map 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 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 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 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 placeholders(String text) { + List 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 000000000000..5a6d7c047220 --- /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. + * + *

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> { + + static final TupleTag> ROUTED = new TupleTag<>(); + static final TupleTag 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 000000000000..c16722f01ae4 --- /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 000000000000..f10e9abc870b --- /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}")); + } +}