ahmedabu98 commented on code in PR #39724: URL: https://github.com/apache/beam/pull/39724#discussion_r3769275974
########## 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(); Review Comment: I think we should keep track of the full map of schemas / partition specs / sort orders. There are instances where we're not always working with the most recent table metadata, so we'll need to have access to historical schema/spec/sort orders. Might make sense to add 3 int fields to indicate the "current" schemaId, specId, orderId, and use them to return the respective metadata property. ########## 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: Also include fileIO json here. Iceberg has a FileIOParser that works similarly to the other properties ########## 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()); + } + + public SideInputTable( + SerializableTableSpec spec, FileIO fileIO, EncryptionManager encryptionManager) { + this.spec = checkNotNull(spec, "spec must not be null"); + this.fileIO = checkNotNull(fileIO, "fileIO must not be null"); + this.encryptionManager = checkNotNull(encryptionManager, "encryptionManager must not be null"); + this.locationProvider = + LocationProviders.locationsFor(spec.getLocation(), spec.getProperties()); + } + + public SerializableTableSpec getTableSpec() { + return spec; + } + + @Override + public String name() { + return spec.getName(); + } + + @Override + public String location() { + return spec.getLocation(); + } + + @Override + public Schema schema() { + return spec.getSchema(); + } + + @Override + public Map<Integer, Schema> schemas() { + return Collections.singletonMap(spec.getSchema().schemaId(), spec.getSchema()); + } + + @Override + public PartitionSpec spec() { + return spec.getPartitionSpec(); + } + + @Override + public Map<Integer, PartitionSpec> specs() { + return Collections.singletonMap(spec.getPartitionSpec().specId(), spec.getPartitionSpec()); + } Review Comment: Following from previous comment. We have instances where we want to access historical partition specs. So this should contain more than just the current spec. Same goes for `schemas()` and `sortOrders()` ########## 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()); + } + + public SideInputTable( + SerializableTableSpec spec, FileIO fileIO, EncryptionManager encryptionManager) { + this.spec = checkNotNull(spec, "spec must not be null"); + this.fileIO = checkNotNull(fileIO, "fileIO must not be null"); + this.encryptionManager = checkNotNull(encryptionManager, "encryptionManager must not be null"); + this.locationProvider = + LocationProviders.locationsFor(spec.getLocation(), spec.getProperties()); + } + + public SerializableTableSpec getTableSpec() { + return spec; + } + + @Override + public String name() { + return spec.getName(); + } + + @Override + public String location() { + return spec.getLocation(); + } + + @Override + public Schema schema() { + return spec.getSchema(); + } + + @Override + public Map<Integer, Schema> schemas() { + return Collections.singletonMap(spec.getSchema().schemaId(), spec.getSchema()); + } + + @Override + public PartitionSpec spec() { + return spec.getPartitionSpec(); + } + + @Override + public Map<Integer, PartitionSpec> specs() { + return Collections.singletonMap(spec.getPartitionSpec().specId(), spec.getPartitionSpec()); + } + + @Override + public SortOrder sortOrder() { + return spec.getSortOrder(); + } + + @Override + public Map<Integer, SortOrder> sortOrders() { + return Collections.singletonMap(spec.getSortOrder().orderId(), spec.getSortOrder()); + } + + @Override + public Map<String, String> properties() { + return spec.getProperties(); + } + + @Override + public LocationProvider locationProvider() { + return locationProvider; + } + + @Override + public FileIO io() { + return fileIO; + } + + @Override + public EncryptionManager encryption() { + return encryptionManager; + } + + @Override + public void refresh() { + // No-op: refresh is managed by the periodic side-input update mechanism Review Comment: Can we throw an `UnsupportedOperationException` for all of these unsupported methods? Better to let us know where we're incorrectly using SerializableTableSpec, and instead opt into using a real table ########## 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: I think we can also serialize properties to reconstruct the EncryptionManager, instead of making it an arg for SideInputTable. Mostly because we won't have access to the EncryptionManager on the worker side to properly create SideInputTable When loading the real table, we can serialize and store the encrypted keys in `SeiralizableTableSpec`: ``` TableMetadata metadata = ((HasTableOperations) table).operations().current(); List<String> encryptedKeyJsons = metadata.encryptionKeys().stream() .map(key -> EncryptedKeyParser.toJson(key, false)) .collect(Collectors.toList()); ``` (AFAICT, we can assume the table implements `HasTableOperations`. Let's throw if not) Then to reconstruct it in `SideInputTable`: ``` List<EncryptedKey> encryptedKeys = encryptedKeyJsons.stream() .map(EncryptedKeyParser::fromJson) .collect(Collectors.toList()); EncryptionManager encryption; if (!properties.containsKey(TableProperties.ENCRYPTION_TABLE_KEY)) { encryption = PlaintextEncryptionManager.instance(); } else { KeyManagementClient kmsClient = EncryptionUtil.createKmsClient(catalogProperties); encryption = EncryptionUtil.createEncryptionManager(encryptedKeys, properties, kmsClient); } return encryption; ``` We'd probably need to pass `catalogProperties` (from IcebergCatalogConfig) to `SideInputTable`'s constructor. ########## 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: Should be able to remove the `FileIO` arg when we serialize it in the tableSpec -- 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]
