jrmccluskey commented on code in PR #39724: URL: https://github.com/apache/beam/pull/39724#discussion_r3813599244
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java: ########## @@ -0,0 +1,219 @@ +/* + * 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.checkStateNotNull; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.Map; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.apache.beam.sdk.schemas.annotations.SchemaIgnore; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.SortOrderParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; + +/** + * A serializable, lightweight representation of an Iceberg {@link Table}'s declarative metadata. + * + * <p>Captures the table's schema, partition spec, sort order, location, properties, and identifier. + * Suitable for broadcasting across worker nodes via Beam's side-input mechanism. + */ +@DefaultSchema(AutoValueSchema.class) +@AutoValue +public abstract class SerializableTableSpec implements Serializable { + + @SchemaFieldNumber("0") + public abstract String getTableIdentifierString(); + + @SchemaFieldNumber("1") + public abstract String getName(); + + @SchemaFieldNumber("2") + public abstract String getLocation(); + + @SchemaFieldNumber("3") + public abstract int getSpecId(); + + @SchemaFieldNumber("4") + public abstract String getSchemaJson(); + + @SchemaFieldNumber("5") + public abstract String getPartitionSpecJson(); + + @SchemaFieldNumber("6") + public abstract String getSortOrderJson(); + + @SchemaFieldNumber("7") + public abstract Map<String, String> getProperties(); + + private transient volatile @MonotonicNonNull Schema cachedSchema; + private transient volatile @MonotonicNonNull PartitionSpec cachedPartitionSpec; + private transient volatile @MonotonicNonNull SortOrder cachedSortOrder; + private transient volatile @MonotonicNonNull TableIdentifier cachedTableIdentifier; + + private static volatile @MonotonicNonNull SchemaCoder<SerializableTableSpec> cachedCoder; + + @SchemaIgnore + public Schema getSchema() { + Schema local = cachedSchema; + if (local == null) { + synchronized (this) { + local = cachedSchema; + if (local == null) { + cachedSchema = local = SchemaParser.fromJson(getSchemaJson()); + } + } + } + return local; + } + + @SchemaIgnore + public PartitionSpec getPartitionSpec() { + PartitionSpec local = cachedPartitionSpec; + if (local == null) { + synchronized (this) { + local = cachedPartitionSpec; + if (local == null) { + cachedPartitionSpec = + local = PartitionSpecParser.fromJson(getSchema(), getPartitionSpecJson()); + } + } + } + return local; + } + + @SchemaIgnore + public SortOrder getSortOrder() { + SortOrder local = cachedSortOrder; + if (local == null) { + synchronized (this) { + local = cachedSortOrder; + if (local == null) { + cachedSortOrder = local = SortOrderParser.fromJson(getSchema(), getSortOrderJson()); + } + } + } + return local; + } + + @SchemaIgnore + public TableIdentifier getTableIdentifier() { + TableIdentifier local = cachedTableIdentifier; + if (local == null) { + synchronized (this) { + local = cachedTableIdentifier; + if (local == null) { + cachedTableIdentifier = + local = IcebergUtils.parseTableIdentifier(getTableIdentifierString()); + } + } + } + return local; + } + + public static Builder builder() { + return new AutoValue_SerializableTableSpec.Builder(); + } + + public abstract Builder toBuilder(); + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setTableIdentifierString(String tableIdentifierString); + + public abstract Builder setName(String name); + + public abstract Builder setLocation(String location); + + public abstract Builder setSpecId(int specId); + + public abstract Builder setSchemaJson(String schemaJson); + + public abstract Builder setPartitionSpecJson(String partitionSpecJson); + + public abstract Builder setSortOrderJson(String sortOrderJson); + + public abstract Builder setProperties(Map<String, String> properties); + + public abstract SerializableTableSpec build(); + } + + /** + * Constructs a {@link SerializableTableSpec} from a {@link Table}, using {@link Table#name()} as + * the table identifier string. + * + * <p>Note: When possible, prefer {@link #fromTable(TableIdentifier, Table)} to avoid catalog name + * prefix ambiguities in {@link Table#name()}. + */ + public static SerializableTableSpec fromTable(Table table) { + return fromTable(table.name(), table); + } + + /** + * Constructs a {@link SerializableTableSpec} from a {@link TableIdentifier} and a {@link Table}. + */ + public static SerializableTableSpec fromTable(TableIdentifier tableIdentifier, Table table) { + return fromTable(IcebergUtils.tableIdentifierToString(tableIdentifier), table); + } + + /** + * Constructs a {@link SerializableTableSpec} from an explicit table identifier string and a + * {@link Table}. + */ + public static SerializableTableSpec fromTable(String tableIdentifierString, Table table) { + return builder() + .setTableIdentifierString(tableIdentifierString) + .setName(table.name()) + .setLocation(table.location()) + .setSpecId(table.spec().specId()) + .setSchemaJson(SchemaParser.toJson(table.schema())) + .setPartitionSpecJson(PartitionSpecParser.toJson(table.spec())) + .setSortOrderJson(SortOrderParser.toJson(table.sortOrder())) + .setProperties(table.properties()) + .build(); + } Review Comment: Done ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java: ########## @@ -0,0 +1,346 @@ +/* + * 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.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.apache.beam.sdk.annotations.Internal; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DeleteFiles; +import org.apache.iceberg.ExpireSnapshots; +import org.apache.iceberg.HistoryEntry; +import org.apache.iceberg.IncrementalAppendScan; +import org.apache.iceberg.IncrementalChangelogScan; +import org.apache.iceberg.LocationProviders; +import org.apache.iceberg.ManageSnapshots; +import org.apache.iceberg.OverwriteFiles; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.ReplacePartitions; +import org.apache.iceberg.ReplaceSortOrder; +import org.apache.iceberg.RewriteFiles; +import org.apache.iceberg.RewriteManifests; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.UpdateLocation; +import org.apache.iceberg.UpdatePartitionSpec; +import org.apache.iceberg.UpdateProperties; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.UpdateStatistics; +import org.apache.iceberg.encryption.EncryptionManager; +import org.apache.iceberg.encryption.PlaintextEncryptionManager; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.LocationProvider; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * A lightweight adapter that implements {@link Table} backed by a {@link SerializableTableSpec}. + * + * <p>Delegates declarative metadata (schema, partition specs, sort order, properties) to the + * broadcasted {@link SerializableTableSpec} and file I/O to a worker-local {@link FileIO} instance. + * + * <p>Mutation operations (e.g. {@code newAppend()}, {@code updateSchema()}) throw {@link + * UnsupportedOperationException} because table commits are handled centrally in {@link + * AppendFilesToTables}. + */ +@Internal +@SuppressWarnings("nullness") +public class SideInputTable implements Table { + + private final SerializableTableSpec spec; + private final FileIO fileIO; + private final EncryptionManager encryptionManager; + private final LocationProvider locationProvider; + + public SideInputTable(SerializableTableSpec spec, FileIO fileIO) { + this(spec, fileIO, PlaintextEncryptionManager.instance()); + } Review Comment: Done -- 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]
