This is an automated email from the ASF dual-hosted git repository.
jrmccluskey 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 a26ecfd4857 [#39723] Implement model for Iceberg side input cache
(#39724)
a26ecfd4857 is described below
commit a26ecfd4857264ea121d0d41929b441d05400862
Author: Jack McCluskey <[email protected]>
AuthorDate: Thu Aug 20 12:40:22 2026 -0400
[#39723] Implement model for Iceberg side input cache (#39724)
* [#39723] Implement model for Iceberg side input cache
* Move FileIO and EncryptionManager, address comments
---
.../beam/sdk/io/iceberg/SerializableTableSpec.java | 382 +++++++++++++++++++++
.../apache/beam/sdk/io/iceberg/SideInputTable.java | 368 ++++++++++++++++++++
.../sdk/io/iceberg/SerializableTableSpecTest.java | 348 +++++++++++++++++++
.../beam/sdk/io/iceberg/SideInputTableTest.java | 239 +++++++++++++
4 files changed, 1337 insertions(+)
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java
new file mode 100644
index 00000000000..c6ee4a97699
--- /dev/null
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java
@@ -0,0 +1,382 @@
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+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.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.EncryptedKeyParser;
+import org.apache.iceberg.HasTableOperations;
+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.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptedKey;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileIOParser;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A serializable, lightweight representation of an Iceberg {@link Table}'s
declarative metadata.
+ *
+ * <p>Captures the table's schemas, partition specs, sort orders, location,
properties, identifier,
+ * encrypted keys, and serialized {@link FileIO} configuration. 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 getSchemaId();
+
+ @SchemaFieldNumber("4")
+ public abstract Map<Integer, String> getSchemasJson();
+
+ @SchemaFieldNumber("5")
+ public abstract int getSpecId();
+
+ @SchemaFieldNumber("6")
+ public abstract Map<Integer, String> getPartitionSpecsJson();
+
+ @SchemaFieldNumber("7")
+ public abstract int getOrderId();
+
+ @SchemaFieldNumber("8")
+ public abstract Map<Integer, String> getSortOrdersJson();
+
+ @SchemaFieldNumber("9")
+ public abstract Map<String, String> getProperties();
+
+ @SchemaFieldNumber("10")
+ public abstract String getFileIoJson();
+
+ @SchemaFieldNumber("11")
+ public abstract List<String> getEncryptedKeyJsons();
+
+ private transient volatile @MonotonicNonNull Map<Integer, Schema>
cachedSchemas;
+ private transient volatile @MonotonicNonNull Map<Integer, PartitionSpec>
cachedPartitionSpecs;
+ private transient volatile @MonotonicNonNull Map<Integer, SortOrder>
cachedSortOrders;
+ private transient volatile @MonotonicNonNull TableIdentifier
cachedTableIdentifier;
+ private transient volatile @MonotonicNonNull FileIO cachedFileIO;
+ private transient volatile @MonotonicNonNull List<EncryptedKey>
cachedEncryptedKeys;
+
+ private static volatile @MonotonicNonNull SchemaCoder<SerializableTableSpec>
cachedCoder;
+
+ @SchemaIgnore
+ public Map<Integer, Schema> getSchemas() {
+ Map<Integer, Schema> local = cachedSchemas;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedSchemas;
+ if (local == null) {
+ ImmutableMap.Builder<Integer, Schema> builder =
ImmutableMap.builder();
+ for (Map.Entry<Integer, String> entry : getSchemasJson().entrySet())
{
+ builder.put(entry.getKey(),
SchemaParser.fromJson(entry.getValue()));
+ }
+ cachedSchemas = local = builder.build();
+ }
+ }
+ }
+ return local;
+ }
+
+ @SchemaIgnore
+ public Schema getSchema() {
+ Schema schema = getSchemas().get(getSchemaId());
+ if (schema == null) {
+ throw new IllegalStateException(
+ "Schema with id " + getSchemaId() + " not found in schemas map");
+ }
+ return schema;
+ }
+
+ @SchemaIgnore
+ public @Nullable Schema getSchema(int schemaId) {
+ return getSchemas().get(schemaId);
+ }
+
+ @SchemaIgnore
+ public Map<Integer, PartitionSpec> getPartitionSpecs() {
+ Map<Integer, PartitionSpec> local = cachedPartitionSpecs;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedPartitionSpecs;
+ if (local == null) {
+ ImmutableMap.Builder<Integer, PartitionSpec> builder =
ImmutableMap.builder();
+ for (Map.Entry<Integer, String> entry :
getPartitionSpecsJson().entrySet()) {
+ builder.put(
+ entry.getKey(), PartitionSpecParser.fromJson(getSchema(),
entry.getValue()));
+ }
+ cachedPartitionSpecs = local = builder.build();
+ }
+ }
+ }
+ return local;
+ }
+
+ @SchemaIgnore
+ public PartitionSpec getPartitionSpec() {
+ PartitionSpec spec = getPartitionSpecs().get(getSpecId());
+ if (spec == null) {
+ throw new IllegalStateException(
+ "PartitionSpec with id " + getSpecId() + " not found in
partitionSpecs map");
+ }
+ return spec;
+ }
+
+ @SchemaIgnore
+ public @Nullable PartitionSpec getPartitionSpec(int specId) {
+ return getPartitionSpecs().get(specId);
+ }
+
+ @SchemaIgnore
+ public Map<Integer, SortOrder> getSortOrders() {
+ Map<Integer, SortOrder> local = cachedSortOrders;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedSortOrders;
+ if (local == null) {
+ ImmutableMap.Builder<Integer, SortOrder> builder =
ImmutableMap.builder();
+ for (Map.Entry<Integer, String> entry :
getSortOrdersJson().entrySet()) {
+ builder.put(entry.getKey(), SortOrderParser.fromJson(getSchema(),
entry.getValue()));
+ }
+ cachedSortOrders = local = builder.build();
+ }
+ }
+ }
+ return local;
+ }
+
+ @SchemaIgnore
+ public SortOrder getSortOrder() {
+ SortOrder order = getSortOrders().get(getOrderId());
+ if (order == null) {
+ throw new IllegalStateException(
+ "SortOrder with id " + getOrderId() + " not found in sortOrders
map");
+ }
+ return order;
+ }
+
+ @SchemaIgnore
+ public @Nullable SortOrder getSortOrder(int orderId) {
+ return getSortOrders().get(orderId);
+ }
+
+ @SchemaIgnore
+ public TableIdentifier getTableIdentifier() {
+ TableIdentifier local = cachedTableIdentifier;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedTableIdentifier;
+ if (local == null) {
+ cachedTableIdentifier =
+ local =
IcebergUtils.parseTableIdentifier(getTableIdentifierString());
+ }
+ }
+ }
+ return local;
+ }
+
+ @SchemaIgnore
+ public FileIO getFileIO() {
+ FileIO local = cachedFileIO;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedFileIO;
+ if (local == null) {
+ cachedFileIO = local = FileIOParser.fromJson(getFileIoJson());
+ }
+ }
+ }
+ return local;
+ }
+
+ @SchemaIgnore
+ public List<EncryptedKey> getEncryptedKeys() {
+ List<EncryptedKey> local = cachedEncryptedKeys;
+ if (local == null) {
+ synchronized (this) {
+ local = cachedEncryptedKeys;
+ if (local == null) {
+ cachedEncryptedKeys =
+ local =
+ getEncryptedKeyJsons().stream()
+ .map(EncryptedKeyParser::fromJson)
+ .collect(Collectors.toList());
+ }
+ }
+ }
+ 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 setSchemaId(int schemaId);
+
+ public abstract Builder setSchemasJson(Map<Integer, String> schemasJson);
+
+ public abstract Builder setSpecId(int specId);
+
+ public abstract Builder setPartitionSpecsJson(Map<Integer, String>
partitionSpecsJson);
+
+ public abstract Builder setOrderId(int orderId);
+
+ public abstract Builder setSortOrdersJson(Map<Integer, String>
sortOrdersJson);
+
+ public abstract Builder setProperties(Map<String, String> properties);
+
+ public abstract Builder setFileIoJson(String fileIoJson);
+
+ public abstract Builder setEncryptedKeyJsons(List<String>
encryptedKeyJsons);
+
+ @SchemaIgnore
+ public Builder setFileIO(FileIO fileIO) {
+ return setFileIoJson(FileIOParser.toJson(fileIO));
+ }
+
+ 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) {
+ if (!(table instanceof HasTableOperations)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Table %s of class %s does not implement HasTableOperations",
+ table.name(), table.getClass().getName()));
+ }
+
+ TableMetadata metadata = ((HasTableOperations)
table).operations().current();
+ List<String> encryptedKeyJsons = Collections.emptyList();
+ if (metadata != null && metadata.encryptionKeys() != null) {
+ encryptedKeyJsons =
+ metadata.encryptionKeys().stream()
+ .map(key -> EncryptedKeyParser.toJson(key, false))
+ .collect(Collectors.toList());
+ }
+
+ ImmutableMap.Builder<Integer, String> schemasJson = ImmutableMap.builder();
+ for (Map.Entry<Integer, Schema> entry : table.schemas().entrySet()) {
+ schemasJson.put(entry.getKey(), SchemaParser.toJson(entry.getValue()));
+ }
+
+ ImmutableMap.Builder<Integer, String> specsJson = ImmutableMap.builder();
+ for (Map.Entry<Integer, PartitionSpec> entry : table.specs().entrySet()) {
+ specsJson.put(entry.getKey(),
PartitionSpecParser.toJson(entry.getValue()));
+ }
+
+ ImmutableMap.Builder<Integer, String> sortOrdersJson =
ImmutableMap.builder();
+ for (Map.Entry<Integer, SortOrder> entry : table.sortOrders().entrySet()) {
+ sortOrdersJson.put(entry.getKey(),
SortOrderParser.toJson(entry.getValue()));
+ }
+
+ return builder()
+ .setTableIdentifierString(tableIdentifierString)
+ .setName(table.name())
+ .setLocation(table.location())
+ .setSchemaId(table.schema().schemaId())
+ .setSchemasJson(schemasJson.build())
+ .setSpecId(table.spec().specId())
+ .setPartitionSpecsJson(specsJson.build())
+ .setOrderId(table.sortOrder().orderId())
+ .setSortOrdersJson(sortOrdersJson.build())
+ .setProperties(table.properties())
+ .setFileIoJson(FileIOParser.toJson(table.io()))
+ .setEncryptedKeyJsons(encryptedKeyJsons)
+ .build();
+ }
+
+ /** Returns the cached {@link SchemaCoder} for {@link
SerializableTableSpec}. */
+ public static SchemaCoder<SerializableTableSpec> getCoder() {
+ if (cachedCoder == null) {
+ synchronized (SerializableTableSpec.class) {
+ if (cachedCoder == null) {
+ try {
+ cachedCoder =
+
SchemaRegistry.createDefault().getSchemaCoder(SerializableTableSpec.class);
+ } catch (NoSuchSchemaException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+ return checkStateNotNull(cachedCoder);
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java
new file mode 100644
index 00000000000..aa51571c4cb
--- /dev/null
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java
@@ -0,0 +1,368 @@
+/*
+ * 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.TableProperties;
+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.EncryptionUtil;
+import org.apache.iceberg.encryption.KeyManagementClient;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+
+/**
+ * A lightweight adapter that implements {@link Table} backed by a {@link
SerializableTableSpec}.
+ *
+ * <p>Delegates declarative metadata (schemas, partition specs, sort orders,
properties) and {@link
+ * FileIO} to the broadcasted {@link SerializableTableSpec}, and reconstructs
the {@link
+ * EncryptionManager} from catalog properties or falls back to {@link
PlaintextEncryptionManager}.
+ *
+ * <p>All non-metadata or mutating operations (e.g. {@code refresh()}, {@code
currentSnapshot()},
+ * {@code newAppend()}, {@code updateSchema()}) throw {@link
UnsupportedOperationException}. Table
+ * commits are handled centrally in {@link AppendFilesToTables}.
+ */
+@Internal
+@SuppressWarnings("nullness")
+public class SideInputTable implements Table {
+
+ private final SerializableTableSpec spec;
+ private final EncryptionManager encryptionManager;
+ private final LocationProvider locationProvider;
+
+ public SideInputTable(SerializableTableSpec spec) {
+ this(spec, Collections.emptyMap());
+ }
+
+ public SideInputTable(SerializableTableSpec spec, Map<String, String>
catalogProperties) {
+ this.spec = checkNotNull(spec, "spec must not be null");
+ checkNotNull(catalogProperties, "catalogProperties must not be null");
+ this.locationProvider =
+ LocationProviders.locationsFor(spec.getLocation(),
spec.getProperties());
+
+ Map<String, String> properties = spec.getProperties();
+ if (!properties.containsKey(TableProperties.ENCRYPTION_TABLE_KEY)) {
+ this.encryptionManager = PlaintextEncryptionManager.instance();
+ } else {
+ KeyManagementClient kmsClient =
EncryptionUtil.createKmsClient(catalogProperties);
+ this.encryptionManager =
+ EncryptionUtil.createEncryptionManager(spec.getEncryptedKeys(),
properties, kmsClient);
+ }
+ }
+
+ public SideInputTable(SerializableTableSpec spec, EncryptionManager
encryptionManager) {
+ this.spec = checkNotNull(spec, "spec 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 spec.getSchemas();
+ }
+
+ @Override
+ public PartitionSpec spec() {
+ return spec.getPartitionSpec();
+ }
+
+ @Override
+ public Map<Integer, PartitionSpec> specs() {
+ return spec.getPartitionSpecs();
+ }
+
+ @Override
+ public SortOrder sortOrder() {
+ return spec.getSortOrder();
+ }
+
+ @Override
+ public Map<Integer, SortOrder> sortOrders() {
+ return spec.getSortOrders();
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return spec.getProperties();
+ }
+
+ @Override
+ public LocationProvider locationProvider() {
+ return locationProvider;
+ }
+
+ @Override
+ public FileIO io() {
+ return spec.getFileIO();
+ }
+
+ @Override
+ public EncryptionManager encryption() {
+ return encryptionManager;
+ }
+
+ @Override
+ public void refresh() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
refresh.");
+ }
+
+ @Override
+ public Snapshot currentSnapshot() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
snapshots.");
+ }
+
+ @Override
+ public Snapshot snapshot(long snapshotId) {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
snapshots.");
+ }
+
+ @Override
+ public Iterable<Snapshot> snapshots() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
snapshots.");
+ }
+
+ @Override
+ public List<HistoryEntry> history() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
snapshots.");
+ }
+
+ @Override
+ public Map<String, SnapshotRef> refs() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
snapshot refs.");
+ }
+
+ @Override
+ public List<StatisticsFile> statisticsFiles() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
statisticsFiles.");
+ }
+
+ @Override
+ public List<PartitionStatisticsFile> partitionStatisticsFiles() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
partitionStatisticsFiles.");
+ }
+
+ @Override
+ public TableScan newScan() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
scans directly.");
+ }
+
+ @Override
+ public IncrementalAppendScan newIncrementalAppendScan() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
scans directly.");
+ }
+
+ @Override
+ public IncrementalChangelogScan newIncrementalChangelogScan() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
scans directly.");
+ }
+
+ @Override
+ public UpdateSchema updateSchema() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public UpdatePartitionSpec updateSpec() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public UpdateProperties updateProperties() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public ReplaceSortOrder replaceSortOrder() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public UpdateLocation updateLocation() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public AppendFiles newAppend() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public AppendFiles newFastAppend() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public RewriteFiles newRewrite() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public RewriteManifests rewriteManifests() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public OverwriteFiles newOverwrite() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public RowDelta newRowDelta() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public ReplacePartitions newReplacePartitions() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public DeleteFiles newDelete() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public UpdateStatistics updateStatistics() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public ExpireSnapshots expireSnapshots() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public ManageSnapshots manageSnapshots() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public Transaction newTransaction() {
+ throw new UnsupportedOperationException(
+ "SideInputTable is a read-only metadata adapter and does not support
table mutations.");
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof SideInputTable)) {
+ return false;
+ }
+ SideInputTable that = (SideInputTable) o;
+ return Objects.equals(spec, that.spec)
+ && Objects.equals(encryptionManager, that.encryptionManager);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(spec, encryptionManager);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("spec", spec)
+ .add("encryptionManager", encryptionManager)
+ .toString();
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java
new file mode 100644
index 00000000000..87a843db7a2
--- /dev/null
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java
@@ -0,0 +1,348 @@
+/*
+ * 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.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.util.CoderUtils;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptedKey;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class SerializableTableSpecTest {
+
+ @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private Catalog catalog;
+ private String warehouseLocation;
+
+ private static final Schema COMPLEX_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "timestamp_val", Types.TimestampType.withZone()),
+ optional(4, "amount", Types.DecimalType.of(10, 2)),
+ optional(
+ 5,
+ "nested_struct",
+ Types.StructType.of(
+ required(6, "nested_id", Types.IntegerType.get()),
+ optional(7, "nested_desc", Types.StringType.get()))),
+ optional(8, "string_list", Types.ListType.ofOptional(9,
Types.StringType.get())),
+ optional(
+ 10,
+ "str_int_map",
+ Types.MapType.ofOptional(11, 12, Types.StringType.get(),
Types.IntegerType.get())));
+
+ @Before
+ public void setUp() throws Exception {
+ warehouseLocation = "file:" + tempFolder.newFolder().getAbsolutePath();
+ catalog =
+ CatalogUtil.loadCatalog(
+ CatalogUtil.ICEBERG_CATALOG_HADOOP,
+ "hadoop",
+ ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
warehouseLocation),
+ new Configuration());
+ }
+
+ @Test
+ public void testFromTableThrowsWhenNotImplementingHasTableOperations() {
+ Table mockTable = mock(Table.class);
+ when(mockTable.name()).thenReturn("mock_table");
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> SerializableTableSpec.fromTable(TableIdentifier.of("default",
"mock"), mockTable));
+ }
+
+ @Test
+ public void testFromTableAndGettersUnpartitioned() {
+ TableIdentifier tableId = TableIdentifier.of("default",
"unpartitioned_table");
+ Table table = catalog.createTable(tableId, TestFixtures.SCHEMA);
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
table);
+
+ assertEquals(IcebergUtils.tableIdentifierToString(tableId),
spec.getTableIdentifierString());
+ assertEquals(table.name(), spec.getName());
+ assertEquals(table.location(), spec.getLocation());
+ assertEquals(table.schema().schemaId(), spec.getSchemaId());
+ assertEquals(table.spec().specId(), spec.getSpecId());
+ assertEquals(table.sortOrder().orderId(), spec.getOrderId());
+ assertEquals(table.schema().asStruct(), spec.getSchema().asStruct());
+ assertEquals(table.schemas().size(), spec.getSchemas().size());
+ assertEquals(table.spec(), spec.getPartitionSpec());
+ assertEquals(table.specs().size(), spec.getPartitionSpecs().size());
+ assertTrue(spec.getPartitionSpec().isUnpartitioned());
+ assertEquals(table.sortOrder(), spec.getSortOrder());
+ assertEquals(table.sortOrders().size(), spec.getSortOrders().size());
+ assertEquals(tableId, spec.getTableIdentifier());
+ assertNotNull(spec.getFileIO());
+ assertEquals(table.io().getClass().getName(),
spec.getFileIO().getClass().getName());
+ assertNotNull(spec.getEncryptedKeyJsons());
+ assertNotNull(spec.getEncryptedKeys());
+ assertTrue(spec.getEncryptedKeys().isEmpty());
+ }
+
+ @Test
+ public void testFromTableAndGettersPartitionedWithSortOrder() {
+ TableIdentifier tableId = TableIdentifier.of("default",
"partitioned_table");
+ PartitionSpec partitionSpec =
+
PartitionSpec.builderFor(COMPLEX_SCHEMA).day("timestamp_val").identity("name").build();
+ SortOrder sortOrder =
+ SortOrder.builderFor(COMPLEX_SCHEMA)
+ .sortBy("id", SortDirection.ASC, NullOrder.NULLS_FIRST)
+ .sortBy("name", SortDirection.DESC, NullOrder.NULLS_LAST)
+ .build();
+ Map<String, String> properties =
+ ImmutableMap.of("write.format.default", "parquet", "custom.property",
"test-val");
+
+ Table table =
+ catalog
+ .buildTable(tableId, COMPLEX_SCHEMA)
+ .withPartitionSpec(partitionSpec)
+ .withSortOrder(sortOrder)
+ .withProperties(properties)
+ .create();
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(table);
+
+ assertEquals(table.name(), spec.getTableIdentifierString());
+ assertEquals(table.name(), spec.getName());
+ assertEquals(table.location(), spec.getLocation());
+ assertEquals(table.schema().schemaId(), spec.getSchemaId());
+ assertEquals(partitionSpec.specId(), spec.getSpecId());
+ assertEquals(sortOrder.orderId(), spec.getOrderId());
+ assertEquals(table.schema().asStruct(), spec.getSchema().asStruct());
+ assertEquals(table.schemas().keySet(), spec.getSchemas().keySet());
+ assertEquals(partitionSpec, spec.getPartitionSpec());
+ assertEquals(table.specs().keySet(), spec.getPartitionSpecs().keySet());
+ assertEquals(sortOrder, spec.getSortOrder());
+ assertEquals(table.sortOrders().keySet(), spec.getSortOrders().keySet());
+ assertEquals(
+ properties.get("write.format.default"),
spec.getProperties().get("write.format.default"));
+ assertEquals(properties.get("custom.property"),
spec.getProperties().get("custom.property"));
+ assertNotNull(spec.getFileIO());
+ assertNotNull(spec.getEncryptedKeys());
+ }
+
+ @Test
+ public void testDottedNestedNamespaceIdentifier() {
+ TableIdentifier tableId = TableIdentifier.of("my", "nested", "catalog",
"deep_table");
+ Table table = catalog.createTable(tableId, TestFixtures.SCHEMA);
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
table);
+
+ assertEquals("my.nested.catalog.deep_table",
spec.getTableIdentifierString());
+ assertEquals(tableId, spec.getTableIdentifier());
+ }
+
+ @Test
+ public void testBuilderAndToBuilderWithEmptyProperties() {
+ TableIdentifier tableId = TableIdentifier.of("default",
"empty_prop_table");
+ Table table = catalog.createTable(tableId, TestFixtures.SCHEMA);
+
+ SerializableTableSpec spec =
+ SerializableTableSpec.fromTable(tableId, table)
+ .toBuilder()
+ .setProperties(Collections.emptyMap())
+ .build();
+
+ assertTrue(spec.getProperties().isEmpty());
+ assertEquals(tableId, spec.getTableIdentifier());
+ assertNotNull(spec.getFileIO());
+ }
+
+ @Test
+ public void testJavaSerializationRoundtrip() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of("default", "ser_table");
+ PartitionSpec partitionSpec =
+ PartitionSpec.builderFor(COMPLEX_SCHEMA).bucket("name", 16).build();
+ Table table =
+ catalog.buildTable(tableId,
COMPLEX_SCHEMA).withPartitionSpec(partitionSpec).create();
+
+ SerializableTableSpec original = SerializableTableSpec.fromTable(tableId,
table);
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+ oos.writeObject(original);
+ }
+
+ SerializableTableSpec deserialized;
+ try (ObjectInputStream ois =
+ new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) {
+ deserialized = (SerializableTableSpec) ois.readObject();
+ }
+
+ assertNotNull(deserialized);
+ assertEquals(original, deserialized);
+ assertEquals(original.getTableIdentifierString(),
deserialized.getTableIdentifierString());
+ assertEquals(original.getSchemaId(), deserialized.getSchemaId());
+ assertEquals(original.getSpecId(), deserialized.getSpecId());
+ assertEquals(original.getOrderId(), deserialized.getOrderId());
+ assertEquals(original.getSchema().asStruct(),
deserialized.getSchema().asStruct());
+ assertEquals(original.getSchemas().keySet(),
deserialized.getSchemas().keySet());
+ assertEquals(original.getPartitionSpec(), deserialized.getPartitionSpec());
+ assertEquals(original.getPartitionSpecs().keySet(),
deserialized.getPartitionSpecs().keySet());
+ assertEquals(original.getSortOrder(), deserialized.getSortOrder());
+ assertEquals(original.getSortOrders().keySet(),
deserialized.getSortOrders().keySet());
+ assertEquals(original.getEncryptedKeyJsons(),
deserialized.getEncryptedKeyJsons());
+ assertEquals(original.getEncryptedKeys(), deserialized.getEncryptedKeys());
+ assertNotNull(deserialized.getFileIO());
+ }
+
+ @Test
+ public void testBeamSchemaCoderRoundtrip() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of("default", "coder_table");
+ PartitionSpec partitionSpec =
+ PartitionSpec.builderFor(COMPLEX_SCHEMA).hour("timestamp_val").build();
+ SortOrder sortOrder =
+ SortOrder.builderFor(COMPLEX_SCHEMA)
+ .sortBy("id", SortDirection.DESC, NullOrder.NULLS_LAST)
+ .build();
+ Table table =
+ catalog
+ .buildTable(tableId, COMPLEX_SCHEMA)
+ .withPartitionSpec(partitionSpec)
+ .withSortOrder(sortOrder)
+ .withProperties(ImmutableMap.of("k1", "v1"))
+ .create();
+
+ SerializableTableSpec original = SerializableTableSpec.fromTable(tableId,
table);
+ SchemaCoder<SerializableTableSpec> coder =
SerializableTableSpec.getCoder();
+
+ SerializableTableSpec decoded = CoderUtils.clone(coder, original);
+
+ assertNotNull(decoded);
+ assertEquals(original, decoded);
+ assertEquals(original.getTableIdentifierString(),
decoded.getTableIdentifierString());
+ assertEquals(original.getName(), decoded.getName());
+ assertEquals(original.getLocation(), decoded.getLocation());
+ assertEquals(original.getSchemaId(), decoded.getSchemaId());
+ assertEquals(original.getSchemasJson(), decoded.getSchemasJson());
+ assertEquals(original.getSpecId(), decoded.getSpecId());
+ assertEquals(original.getPartitionSpecsJson(),
decoded.getPartitionSpecsJson());
+ assertEquals(original.getOrderId(), decoded.getOrderId());
+ assertEquals(original.getSortOrdersJson(), decoded.getSortOrdersJson());
+ assertEquals(original.getSchema().asStruct(),
decoded.getSchema().asStruct());
+ assertEquals(original.getPartitionSpec(), decoded.getPartitionSpec());
+ assertEquals(original.getSortOrder(), decoded.getSortOrder());
+ assertEquals(original.getProperties(), decoded.getProperties());
+ assertEquals(original.getFileIoJson(), decoded.getFileIoJson());
+ assertEquals(original.getEncryptedKeyJsons(),
decoded.getEncryptedKeyJsons());
+ assertNotNull(decoded.getFileIO());
+ }
+
+ @Test
+ @SuppressWarnings("ReferenceEquality")
+ public void testConcurrentGetterInitializationThreadSafety() throws
Exception {
+ TableIdentifier tableId = TableIdentifier.of("default",
"concurrent_table");
+ PartitionSpec partitionSpec =
+ PartitionSpec.builderFor(COMPLEX_SCHEMA).bucket("name", 8).build();
+ Table table =
+ catalog.buildTable(tableId,
COMPLEX_SCHEMA).withPartitionSpec(partitionSpec).create();
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
table);
+
+ int numThreads = 16;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ try {
+ for (int i = 0; i < numThreads; i++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ startLatch.await();
+ Schema schema = spec.getSchema();
+ Map<Integer, Schema> schemas = spec.getSchemas();
+ PartitionSpec ps = spec.getPartitionSpec();
+ Map<Integer, PartitionSpec> specs = spec.getPartitionSpecs();
+ SortOrder so = spec.getSortOrder();
+ Map<Integer, SortOrder> orders = spec.getSortOrders();
+ TableIdentifier ti = spec.getTableIdentifier();
+ FileIO io = spec.getFileIO();
+ List<EncryptedKey> keys = spec.getEncryptedKeys();
+
+ if (schema == null
+ || schemas == null
+ || ps == null
+ || specs == null
+ || so == null
+ || orders == null
+ || ti == null
+ || io == null
+ || keys == null) {
+ throw new IllegalStateException("Getter returned null");
+ }
+ if (schemas != spec.getSchemas()
+ || specs != spec.getPartitionSpecs()
+ || orders != spec.getSortOrders()
+ || io != spec.getFileIO()
+ || keys != spec.getEncryptedKeys()) {
+ throw new IllegalStateException("Getter returned
non-identical instance");
+ }
+ return null;
+ }));
+ }
+
+ startLatch.countDown();
+ for (Future<Void> future : futures) {
+ future.get(10, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdown();
+ }
+ }
+}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java
new file mode 100644
index 00000000000..663c818b587
--- /dev/null
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java
@@ -0,0 +1,239 @@
+/*
+ * 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.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Map;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.NullOrder;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.SortDirection;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.encryption.PlaintextEncryptionManager;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class SideInputTableTest {
+
+ @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private Catalog catalog;
+ private String warehouseLocation;
+
+ @Before
+ public void setUp() throws Exception {
+ warehouseLocation = "file:" + tempFolder.newFolder().getAbsolutePath();
+ catalog =
+ CatalogUtil.loadCatalog(
+ CatalogUtil.ICEBERG_CATALOG_HADOOP,
+ "hadoop",
+ ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
warehouseLocation),
+ new Configuration());
+ }
+
+ @Test
+ public void testConstructorNullChecks() {
+ assertThrows(NullPointerException.class, () -> new SideInputTable(null));
+ TableIdentifier tableId = TableIdentifier.of("default",
"null_check_table");
+ Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA);
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
realTable);
+ assertThrows(
+ NullPointerException.class, () -> new SideInputTable(spec,
(Map<String, String>) null));
+ }
+
+ @Test
+ public void testMetadataDelegation() {
+ TableIdentifier tableId = TableIdentifier.of("default",
"side_input_test_table");
+ PartitionSpec partitionSpec =
+ PartitionSpec.builderFor(TestFixtures.SCHEMA).identity("data").build();
+ SortOrder sortOrder =
+ SortOrder.builderFor(TestFixtures.SCHEMA)
+ .sortBy("id", SortDirection.ASC, NullOrder.NULLS_FIRST)
+ .build();
+ Map<String, String> properties =
+ ImmutableMap.of("write.format.default", "parquet", "user.key",
"user.val");
+
+ Table realTable =
+ catalog
+ .buildTable(tableId, TestFixtures.SCHEMA)
+ .withPartitionSpec(partitionSpec)
+ .withSortOrder(sortOrder)
+ .withProperties(properties)
+ .create();
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
realTable);
+ SideInputTable sideInputTable = new SideInputTable(spec,
ImmutableMap.of());
+
+ assertEquals(realTable.name(), sideInputTable.name());
+ assertEquals(realTable.location(), sideInputTable.location());
+ assertEquals(realTable.schema().asStruct(),
sideInputTable.schema().asStruct());
+ assertEquals(realTable.schemas().keySet(),
sideInputTable.schemas().keySet());
+ assertEquals(realTable.spec(), sideInputTable.spec());
+ assertEquals(realTable.specs().keySet(), sideInputTable.specs().keySet());
+ assertEquals(realTable.sortOrder(), sideInputTable.sortOrder());
+ assertEquals(realTable.sortOrders().keySet(),
sideInputTable.sortOrders().keySet());
+ assertEquals(
+ realTable.properties().get("user.key"),
sideInputTable.properties().get("user.key"));
+ assertNotNull(sideInputTable.io());
+ assertEquals(realTable.io().getClass().getName(),
sideInputTable.io().getClass().getName());
+ assertNotNull(sideInputTable.locationProvider());
+ assertNotNull(sideInputTable.encryption());
+ assertTrue(sideInputTable.encryption() instanceof
PlaintextEncryptionManager);
+ assertEquals(spec, sideInputTable.getTableSpec());
+ assertTrue(sideInputTable.specs().containsKey(spec.getSpecId()));
+ }
+
+ @Test
+ public void testWritingPartitionedWithRecordWriter() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of("default",
"partitioned_writer_table");
+ PartitionSpec partitionSpec =
+ PartitionSpec.builderFor(TestFixtures.SCHEMA).identity("data").build();
+ Table realTable =
+ catalog.buildTable(tableId,
TestFixtures.SCHEMA).withPartitionSpec(partitionSpec).create();
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
realTable);
+ SideInputTable sideInputTable = new SideInputTable(spec);
+
+ PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(),
sideInputTable.schema());
+ Record record = GenericRecord.create(sideInputTable.schema());
+ record.setField("id", 42L);
+ record.setField("data", "test_partition_value");
+ partitionKey.partition(record);
+
+ RecordWriter writer =
+ new RecordWriter(
+ sideInputTable, FileFormat.PARQUET, "test_file_001", partitionKey,
ImmutableMap.of());
+
+ writer.write(record);
+ writer.close();
+
+ assertNotNull(writer.getDataFile());
+ assertNotNull(writer.getDataFile().path());
+ assertEquals(1, writer.getDataFile().recordCount());
+ assertEquals(FileFormat.PARQUET, writer.getDataFile().format());
+ }
+
+ @Test
+ public void testWritingUnpartitionedWithRecordWriter() throws Exception {
+ TableIdentifier tableId = TableIdentifier.of("default",
"unpartitioned_writer_table");
+ Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA);
+
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
realTable);
+ SideInputTable sideInputTable = new SideInputTable(spec);
+
+ PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(),
sideInputTable.schema());
+ Record record = GenericRecord.create(sideInputTable.schema());
+ record.setField("id", 99L);
+ record.setField("data", "unpartitioned_data");
+ partitionKey.partition(record);
+
+ RecordWriter writer =
+ new RecordWriter(
+ sideInputTable,
+ FileFormat.PARQUET,
+ "test_unpartitioned_file_001",
+ partitionKey,
+ ImmutableMap.of());
+
+ writer.write(record);
+ writer.close();
+
+ assertNotNull(writer.getDataFile());
+ assertNotNull(writer.getDataFile().path());
+ assertEquals(1, writer.getDataFile().recordCount());
+ assertEquals(FileFormat.PARQUET, writer.getDataFile().format());
+ }
+
+ @Test
+ public void testUnsupportedAndNoOpOperationsThrowExceptions() {
+ TableIdentifier tableId = TableIdentifier.of("default",
"mutations_test_table");
+ Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA);
+ SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId,
realTable);
+ SideInputTable sideInputTable = new SideInputTable(spec);
+
+ // Refresh & Snapshot operations must throw UnsupportedOperationException
+ assertThrows(UnsupportedOperationException.class, sideInputTable::refresh);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::currentSnapshot);
+ assertThrows(UnsupportedOperationException.class, () ->
sideInputTable.snapshot(12345L));
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::snapshots);
+ assertThrows(UnsupportedOperationException.class, sideInputTable::history);
+ assertThrows(UnsupportedOperationException.class, sideInputTable::refs);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::statisticsFiles);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::partitionStatisticsFiles);
+
+ // Scans & Mutations
+ assertThrows(UnsupportedOperationException.class, sideInputTable::newScan);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newIncrementalAppendScan);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newIncrementalChangelogScan);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::updateSchema);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::updateSpec);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::updateProperties);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::replaceSortOrder);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::updateLocation);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newAppend);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newFastAppend);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newRewrite);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::rewriteManifests);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newOverwrite);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newRowDelta);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newReplacePartitions);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newDelete);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::updateStatistics);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::expireSnapshots);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::manageSnapshots);
+ assertThrows(UnsupportedOperationException.class,
sideInputTable::newTransaction);
+ }
+
+ @Test
+ public void testEqualsHashCodeAndToString() {
+ TableIdentifier tableId1 = TableIdentifier.of("default", "t1");
+ TableIdentifier tableId2 = TableIdentifier.of("default", "t2");
+ Table realTable1 = catalog.createTable(tableId1, TestFixtures.SCHEMA);
+ Table realTable2 = catalog.createTable(tableId2, TestFixtures.SCHEMA);
+
+ SerializableTableSpec spec1 = SerializableTableSpec.fromTable(tableId1,
realTable1);
+ SerializableTableSpec spec2 = SerializableTableSpec.fromTable(tableId2,
realTable2);
+
+ SideInputTable table1a = new SideInputTable(spec1);
+ SideInputTable table1b = new SideInputTable(spec1);
+ SideInputTable table2 = new SideInputTable(spec2);
+
+ assertEquals(table1a, table1b);
+ assertEquals(table1a.hashCode(), table1b.hashCode());
+ assertNotEquals(table1a, table2);
+ assertNotNull(table1a.toString());
+ assertTrue(table1a.toString().contains("SideInputTable"));
+ }
+}
