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 000000000000..7d2501c65381 --- /dev/null +++ b/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. + * + *

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 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 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 properties); + + public abstract SerializableTableSpec build(); + } + + /** + * Constructs a {@link SerializableTableSpec} from a {@link Table}, using {@link Table#name()} as + * the table identifier string. + * + *

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(); + } + + /** Returns the cached {@link SchemaCoder} for {@link SerializableTableSpec}. */ + public static SchemaCoder 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 000000000000..c318940896fa --- /dev/null +++ b/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}. + * + *

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

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 schemas() { + return Collections.singletonMap(spec.getSchema().schemaId(), spec.getSchema()); + } + + @Override + public PartitionSpec spec() { + return spec.getPartitionSpec(); + } + + @Override + public Map specs() { + return Collections.singletonMap(spec.getPartitionSpec().specId(), spec.getPartitionSpec()); + } + + @Override + public SortOrder sortOrder() { + return spec.getSortOrder(); + } + + @Override + public Map sortOrders() { + return Collections.singletonMap(spec.getSortOrder().orderId(), spec.getSortOrder()); + } + + @Override + public Map 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 + } + + @Override + public @Nullable Snapshot currentSnapshot() { + return null; + } + + @Override + public @Nullable Snapshot snapshot(long snapshotId) { + return null; + } + + @Override + public Iterable snapshots() { + return Collections.emptyList(); + } + + @Override + public List history() { + return Collections.emptyList(); + } + + @Override + public Map refs() { + return Collections.emptyMap(); + } + + @Override + public List statisticsFiles() { + return Collections.emptyList(); + } + + @Override + public List partitionStatisticsFiles() { + return Collections.emptyList(); + } + + @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(fileIO, that.fileIO) + && Objects.equals(encryptionManager, that.encryptionManager); + } + + @Override + public int hashCode() { + return Objects.hash(spec, fileIO, encryptionManager); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("spec", spec) + .add("fileIO", fileIO) + .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 000000000000..8eef7ae30dd7 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java @@ -0,0 +1,282 @@ +/* + * 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.assertTrue; + +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.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 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.spec().specId(), spec.getSpecId()); + assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(table.spec(), spec.getPartitionSpec()); + assertTrue(spec.getPartitionSpec().isUnpartitioned()); + assertEquals(table.sortOrder(), spec.getSortOrder()); + assertEquals(tableId, spec.getTableIdentifier()); + } + + @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 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(partitionSpec.specId(), spec.getSpecId()); + assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(partitionSpec, spec.getPartitionSpec()); + assertEquals(sortOrder, spec.getSortOrder()); + assertEquals( + properties.get("write.format.default"), spec.getProperties().get("write.format.default")); + assertEquals(properties.get("custom.property"), spec.getProperties().get("custom.property")); + } + + @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()); + } + + @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.getSchema().asStruct(), deserialized.getSchema().asStruct()); + assertEquals(original.getPartitionSpec(), deserialized.getPartitionSpec()); + assertEquals(original.getSortOrder(), deserialized.getSortOrder()); + } + + @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 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.getSpecId(), decoded.getSpecId()); + assertEquals(original.getSchema().asStruct(), decoded.getSchema().asStruct()); + assertEquals(original.getPartitionSpec(), decoded.getPartitionSpec()); + assertEquals(original.getSortOrder(), decoded.getSortOrder()); + assertEquals(original.getProperties(), decoded.getProperties()); + } + + @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> futures = new ArrayList<>(); + + try { + for (int i = 0; i < numThreads; i++) { + futures.add( + executor.submit( + () -> { + startLatch.await(); + Schema schema = spec.getSchema(); + PartitionSpec ps = spec.getPartitionSpec(); + SortOrder so = spec.getSortOrder(); + TableIdentifier ti = spec.getTableIdentifier(); + + if (schema == null || ps == null || so == null || ti == null) { + throw new IllegalStateException("Getter returned null"); + } + if (schema != spec.getSchema() || ps != spec.getPartitionSpec()) { + throw new IllegalStateException("Getter returned non-identical instance"); + } + return null; + })); + } + + startLatch.countDown(); + for (Future 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 000000000000..70da317051c6 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java @@ -0,0 +1,250 @@ +/* + * 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.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.util.Collections; +import java.util.Map; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +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() { + 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(null, realTable.io())); + assertThrows(NullPointerException.class, () -> new SideInputTable(spec, null)); + assertThrows(NullPointerException.class, () -> new SideInputTable(spec, realTable.io(), 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 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, realTable.io(), PlaintextEncryptionManager.instance()); + + 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")); + assertEquals(realTable.io(), sideInputTable.io()); + assertNotNull(sideInputTable.locationProvider()); + assertNotNull(sideInputTable.encryption()); + assertEquals(spec, sideInputTable.getTableSpec()); + assertTrue(sideInputTable.specs().containsKey(spec.getPartitionSpec().specId())); + + // Verify refresh is a safe no-op + sideInputTable.refresh(); + } + + @Test + public void testSnapshotQueriesReturnEmptyOrNull() { + TableIdentifier tableId = TableIdentifier.of("default", "snapshot_query_table"); + Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + + assertNull(sideInputTable.currentSnapshot()); + assertNull(sideInputTable.snapshot(12345L)); + assertEquals(Collections.emptyList(), ImmutableList.copyOf(sideInputTable.snapshots())); + assertEquals(Collections.emptyList(), sideInputTable.history()); + assertEquals(Collections.emptyMap(), sideInputTable.refs()); + assertEquals(Collections.emptyList(), sideInputTable.statisticsFiles()); + assertEquals(Collections.emptyList(), sideInputTable.partitionStatisticsFiles()); + } + + @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, realTable.io()); + + 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, realTable.io()); + + 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 testUnsupportedOperationsThrowExceptions() { + 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, realTable.io()); + + 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, realTable1.io()); + SideInputTable table1b = new SideInputTable(spec1, realTable1.io()); + SideInputTable table2 = new SideInputTable(spec2, realTable2.io()); + + assertEquals(table1a, table1b); + assertEquals(table1a.hashCode(), table1b.hashCode()); + assertNotEquals(table1a, table2); + assertNotNull(table1a.toString()); + assertTrue(table1a.toString().contains("SideInputTable")); + } +}