ahmedabu98 commented on code in PR #32366: URL: https://github.com/apache/beam/pull/32366#discussion_r1765720644
########## sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowFilter.java: ########## @@ -0,0 +1,478 @@ +/* + * 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.util; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * A utility that filters fields from Beam {@link Row}s. This filter can be configured to indicate + * what fields you would like to either <strong>keep</strong>, <strong>drop</strong>, or + * <strong>unnest</strong>. Afterward, call {@link #filter(Row)} on a Schema-compatible Row to + * filter it. An un-configured filter will simply return the input row untouched. + * + * <p>Nested fields can be expressed using dot-notation (e.g. {@code "top.middle.nested"}). + * + * <p>Note: You can only specify nested fields when the filter is configured to + * <strong>unnest</strong>. + * + * <p>A configured {@link RowFilter} will naturally produce {@link Row}s with a new Beam {@link + * Schema}. You can access this new Schema via the filter's {@link #outputSchema()}. + * + * <p>Configure a {@link RowFilter} as follows: + * + * <pre>{@code + * // this is an un-configured filter + * RowFilter unconfigured = new RowFilter(beamSchema); + * + * // this filter will exclusively keep these fields and drop everything else + * List<String> fields = Arrays.asList("foo", "bar", "baz"); + * RowFilter keepingFilter = new RowFilter(beamSchema).keeping(fields); + * + * // this filter will drop these fields + * RowFilter droppingFilter = new RowFilter(beamSchema).dropping(fields); + * + * // this filter will unnest fields "abc" and "xyz" to the top level + * List<String> fields = Arrays.asList("foo.abc", "bar.baz.xyz"); + * RowFilter unnestingFilter = new RowFilter(beamSchema).unnesting(fields); + * + * // produces a filtered row + * Row outputRow = keepingFilter.filter(row); + * }</pre> + * + * Check the documentation for {@link #keeping(List)}, {@link #dropping(List)}, and {@link + * #unnesting(List)} for further details on what an output Row can look like. + */ +public class RowFilter implements Serializable { + private final Schema rowSchema; + private @Nullable Schema transformedSchema; + + public RowFilter(Schema rowSchema) { + this.rowSchema = rowSchema; + } + + /** + * Configures this {@link RowFilter} to filter {@link Row}s by keeping only the specified fields. + * Nested fields can be specified using dot-notation. + * + * <p>For example, if we want to keep the list of fields {@code ["foo", "baz"]}, for the input + * {@link Row}: + * + * <pre>{@code + * foo: 123 + * bar: 456 + * baz: + * nested_1: abc + * nested_2: xyz + * }</pre> + * + * we will get the following output {@link Row}: + * + * <pre>{@code + * foo: 123 + * baz + * nested_1: abc + * nested_2: xyz + * }</pre> + */ + public RowFilter keeping(List<String> fields) { + checkUnconfigured(); + verifyNoNestedFields(fields, "keep"); + validateSchemaContainsFields(rowSchema, fields, "\"keep\""); + transformedSchema = keepFields(rowSchema, fields); + return this; + } + + /** + * Configures this {@link RowFilter} to filter {@link Row} by removing the specified fields. + * Nested fields can be specified using dot-notation. + * + * <p>For example, if we want to drop the list of fields {@code ["foo", "baz"]}, for this input + * {@link Row}: + * + * <pre>{@code + * foo: 123 + * bar: 456 + * baz: + * nested_1: abc + * nested_2: xyz + * }</pre> + * + * we will get the following output {@link Row}: + * + * <pre>{@code + * bar: 456 + * }</pre> + */ + public RowFilter dropping(List<String> fields) { + checkUnconfigured(); + verifyNoNestedFields(fields, "drop"); + validateSchemaContainsFields(rowSchema, fields, "\"drop\""); + transformedSchema = dropFields(rowSchema, fields); + return this; + } + + /** + * Configures this {@link RowFilter} to unnest the specified fields to the top-level and keeping + * their leaf names. The unnested fields are kept and everything else is dropped. This will fail + * if two fields have identical leaf names. Nested fields can be specified using dot-notation. + * + * <p>For example, if we want to unnest the list of fields {@code ["abc", "foo.bar", + * "foo.xyz.baz"]}, for this input {@link Row}: + * + * <pre>{@code + * abc: 123 + * foo: Review Comment: P.S. should we also limit `unnest` to top-level fields? ie. in your example one can do `unnest: foo` but not `unnest: foo.xyz` -- 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]
