diff --git a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json index 9a4fbff0c0b7..a7c9525dc7e2 100644 --- a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 2, + "modification": 4, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" } diff --git a/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json b/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json index e3d6056a5de9..c537844dc84a 100644 --- a/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json +++ b/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 1 + "modification": 3 } diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateCDCReadTasksDoFn.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateCDCReadTasksDoFn.java index 9a0049a8651e..a6dfc01c811a 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateCDCReadTasksDoFn.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateCDCReadTasksDoFn.java @@ -31,6 +31,7 @@ import io.delta.kernel.internal.TableImpl; import io.delta.kernel.internal.actions.AddCDCFile; import io.delta.kernel.internal.actions.AddFile; +import io.delta.kernel.internal.actions.Metadata; import io.delta.kernel.internal.util.VectorUtils; import io.delta.kernel.utils.CloseableIterator; import java.time.Instant; @@ -108,18 +109,28 @@ public void processElement(@Element String tablePath, OutputReceiver actionSet = new HashSet<>(); actionSet.add(DeltaAction.CDC); actionSet.add(DeltaAction.ADD); + actionSet.add(DeltaAction.METADATA); // 4. Iterate over commits in the range and group actions by version try (CloseableIterator batchIter = @@ -144,11 +157,26 @@ public void processElement(@Element String tablePath, OutputReceiver= 0 && !batch.getColumnVector(metadataIdx).isNullAt(i)) { + io.delta.kernel.types.StructType schemaAtVersion = + Metadata.fromColumnVector(batch.getColumnVector(metadataIdx), i).getSchema(); + DeltaSchemaUtils.checkReadCompatible( + schemaAtVersion, version, endSchema, resolvedStartVersion, resolvedEndVersion); + } + CommitActionsInfo info = commitActionsMap.computeIfAbsent( version, k -> new CommitActionsInfo(version, timestamp)); diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java index c63fdcacc1c8..b745d13160e5 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java @@ -346,6 +346,18 @@ public PCollection expand(PBegin input) { TableImpl tableImpl = (TableImpl) table; + long resolvedStartVersion; + Long startVersionVal = getStartVersion(); + String startTimestampVal = getStartTimestamp(); + if (startVersionVal != null) { + resolvedStartVersion = startVersionVal; + } else if (startTimestampVal != null) { + long startMillis = java.time.Instant.parse(startTimestampVal).toEpochMilli(); + resolvedStartVersion = tableImpl.getVersionAtOrAfterTimestamp(engine, startMillis); + } else { + throw new IllegalArgumentException("Either startVersion or startTimestamp must be set."); + } + long resolvedEndVersion; Long endVersionVal = getEndVersion(); String endTimestampVal = getEndTimestamp(); @@ -358,11 +370,34 @@ public PCollection expand(PBegin input) { resolvedEndVersion = table.getLatestSnapshot(engine).getVersion(); } - Snapshot endSnapshot = table.getSnapshotAsOfVersion(engine, resolvedEndVersion); + if (resolvedStartVersion > resolvedEndVersion) { + throw new IllegalArgumentException( + String.format( + "Resolved start version %d is greater than resolved end version %d", + resolvedStartVersion, resolvedEndVersion)); + } + + Snapshot startSnapshot = table.getSnapshotAsOfVersion(engine, resolvedStartVersion); + Snapshot endSnapshot = + resolvedStartVersion == resolvedEndVersion + ? startSnapshot + : table.getSnapshotAsOfVersion(engine, resolvedEndVersion); + StructType startDeltaSchema = startSnapshot.getSchema(); StructType deltaSchema = endSnapshot.getSchema(); - if (deltaSchema == null) { + if (startDeltaSchema == null || deltaSchema == null) { throw new IllegalStateException("Table schema is null."); } + // All changes within the range are read using the table schema at the end version, which is + // only supported if the range only spans additive schema changes (see DeltaSchemaUtils). Fail + // early here if the schema at the start version can't be read using the schema at the end + // version. CreateCDCReadTasksDoFn additionally verifies all schema changes within the range + // while reading the commits of the range. + DeltaSchemaUtils.checkReadCompatible( + startDeltaSchema, + resolvedStartVersion, + deltaSchema, + resolvedStartVersion, + resolvedEndVersion); Schema baseSchema = ReadRows.convertToBeamSchema(deltaSchema); Schema publicBeamSchema = buildPublicBeamSchema(baseSchema, getMetadataColumns()); diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtils.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtils.java new file mode 100644 index 000000000000..98259645799e --- /dev/null +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtils.java @@ -0,0 +1,188 @@ +/* + * 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.delta; + +import io.delta.kernel.internal.util.ColumnMapping; +import io.delta.kernel.types.ArrayType; +import io.delta.kernel.types.DataType; +import io.delta.kernel.types.MapType; +import io.delta.kernel.types.StructField; +import io.delta.kernel.types.StructType; + +/** + * Utilities for validating the schema changes of a Delta table within the range of table versions + * that is read by {@link DeltaIO.ReadChanges}. + * + *

All changes within the range are read using the table schema at the end version of the range. + * Similar to change data feed reads in Delta Lake, this is only supported if the range only spans + * additive schema changes, i.e. if data written with the table schema at any version within the + * range can be read using the table schema at the end version (see {@link #isReadCompatible}). + */ +final class DeltaSchemaUtils { + + private DeltaSchemaUtils() {} + + /** + * Throws an {@link IllegalArgumentException} if data written with {@code schema}, the table + * schema at version {@code schemaVersion}, can't be read using {@code endSchema}, the table + * schema at the end version of the range of versions from {@code startVersion} to {@code + * endVersion} that is read. + */ + static void checkReadCompatible( + StructType schema, + long schemaVersion, + StructType endSchema, + long startVersion, + long endVersion) { + if (!isReadCompatible(schema, endSchema)) { + throw new IllegalArgumentException( + String.format( + "Cannot read changes between versions %d and %d since the table schema at version %d " + + "is not compatible with the table schema at the end version %d, which is used " + + "to read all changes. Only additive schema changes, such as adding nullable " + + "columns, are supported within the range of versions that is read. " + + "Schema at version %d: %s, schema at version %d: %s.", + startVersion, + endVersion, + schemaVersion, + endVersion, + schemaVersion, + schema, + endVersion, + endSchema)); + } + } + + /** + * Returns whether data written with {@code dataSchema} can be read using {@code readSchema}, i.e. + * whether {@code readSchema} only differs from {@code dataSchema} by additive changes. + * + *

This mirrors the schema compatibility check of change data feed reads in Delta Lake. The + * following changes are allowed: + * + *

    + *
  • Adding nullable fields, including nested fields. Records read from data written with + * {@code dataSchema} get null values for these fields. + *
  • Relaxing the nullability of a field from non-nullable to nullable. + *
  • Reordering fields, since fields are matched by name. + *
  • Changing field metadata other than column mapping metadata (see {@code + * hasSameColumnMapping()}). For example, column comments, column default values, or + * identity column high-water marks (which writers update whenever they insert rows into the + * table) may change. Such metadata does not affect the data or the Beam schema of the + * emitted rows. + *
+ * + *

All other changes are not allowed. In particular, dropping or renaming fields (including + * dropping and re-adding a field with the same name), changing data types, tightening the + * nullability of fields, and adding non-nullable fields. + */ + static boolean isReadCompatible(StructType dataSchema, StructType readSchema) { + // Every field of the data schema must still exist in the read schema. Fields are matched by + // their (case-sensitive) names since the names are part of the Beam schema of the emitted rows. + // The positions of fields don't matter since data is read from the columns of data files by + // name. + for (StructField dataField : dataSchema.fields()) { + int readFieldIndex = readSchema.indexOf(dataField.getName()); + if (readFieldIndex < 0) { + // The field was dropped or renamed. + return false; + } + StructField readField = readSchema.at(readFieldIndex); + if (!hasSameColumnMapping(dataField, readField) + || !isNullabilityCompatible(dataField.isNullable(), readField.isNullable()) + || !isDataTypeReadCompatible(dataField.getDataType(), readField.getDataType())) { + return false; + } + } + // Fields that only exist in the read schema were added. Data written with the data schema + // doesn't contain values for them, so they are read as nulls and hence must be nullable. + for (StructField readField : readSchema.fields()) { + if (dataSchema.indexOf(readField.getName()) < 0 && !readField.isNullable()) { + return false; + } + } + return true; + } + + /** Returns whether data of type {@code dataType} can be read as type {@code readType}. */ + private static boolean isDataTypeReadCompatible(DataType dataType, DataType readType) { + if (dataType instanceof StructType && readType instanceof StructType) { + return isReadCompatible((StructType) dataType, (StructType) readType); + } else if (dataType instanceof ArrayType && readType instanceof ArrayType) { + // The nullability of array elements is part of the array type, not of field metadata. + ArrayType dataArray = (ArrayType) dataType; + ArrayType readArray = (ArrayType) readType; + return isNullabilityCompatible(dataArray.containsNull(), readArray.containsNull()) + && isDataTypeReadCompatible(dataArray.getElementType(), readArray.getElementType()); + } else if (dataType instanceof MapType && readType instanceof MapType) { + // The nullability of map values is part of the map type, not of field metadata. + MapType dataMap = (MapType) dataType; + MapType readMap = (MapType) readType; + return isNullabilityCompatible(dataMap.isValueContainsNull(), readMap.isValueContainsNull()) + && isDataTypeReadCompatible(dataMap.getKeyType(), readMap.getKeyType()) + && isDataTypeReadCompatible(dataMap.getValueType(), readMap.getValueType()); + } else { + // Primitive types must be the same since type changes (such as widening an integer to a + // long) are not supported. Primitive types don't contain fields, and hence don't carry field + // metadata, so they can be compared directly (for example, this compares the precision and + // scale of decimal types). This also returns false if the two types are of different kinds + // (e.g. a struct and an array). + return dataType.equals(readType); + } + } + + /** + * Returns whether data with nullability {@code dataNullable} can be read with nullability {@code + * readNullable}. Nullability may be relaxed but not tightened, since data written with a nullable + * type may contain null values. + */ + private static boolean isNullabilityCompatible(boolean dataNullable, boolean readNullable) { + return readNullable || !dataNullable; + } + + /** + * Returns whether two fields are read from the same column of the data files. + * + *

If column mapping is enabled for a table, data files identify columns by the physical names + * (and, in the 'id' column mapping mode, by the IDs) stored in the column mapping metadata of the + * fields, instead of by their logical names. Although this is field metadata, it has to be + * compared since all data files in the range of versions that is read are read using the schema + * of the end version. For example, if a column was dropped and then re-added with the same name + * and type, the logical schemas would match. However, data files written before the drop don't + * contain the re-added column, so change records read from those files would silently get null + * values for it. + */ + private static boolean hasSameColumnMapping(StructField left, StructField right) { + // getPhysicalName() returns the logical name of a field that doesn't have a physical name, + // which is the case if column mapping is not enabled. Hence, enabling column mapping on an + // existing table, which sets the physical names of the existing columns to their logical + // names, is not considered to be a change. + if (!ColumnMapping.getPhysicalName(left).equals(ColumnMapping.getPhysicalName(right))) { + return false; + } + // Column IDs are never reused, so fields with different IDs belong to different columns (for + // example, a column that was dropped and re-added) even if their physical names match. IDs are + // only compared if both fields have one, since enabling column mapping on an existing table + // assigns IDs to the existing columns. + String idKey = ColumnMapping.COLUMN_MAPPING_ID_KEY; + if (left.getMetadata().contains(idKey) && right.getMetadata().contains(idKey)) { + return ColumnMapping.getColumnId(left) == ColumnMapping.getColumnId(right); + } + return true; + } +} diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java index 342c4b94761a..d27c31d43e1f 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java @@ -25,6 +25,7 @@ import io.delta.kernel.types.BooleanType; import io.delta.kernel.types.DateType; import io.delta.kernel.types.DoubleType; +import io.delta.kernel.types.FieldMetadata; import io.delta.kernel.types.FloatType; import io.delta.kernel.types.IntegerType; import io.delta.kernel.types.LongType; @@ -1794,6 +1795,202 @@ public void testReadChangesPartialRange() throws Exception { readPipeline.run().waitUntilFinish(); } + @Test + public void testReadChangesNonAdditiveSchemaChangeBetweenVersionsThrows() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-schema-evolution-versions"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // Version 0 creates the table with schema (name, age) and version 1 drops the column 'age'. + DeltaWriteTestUtils.setupDroppedColumnTable(engine, tableDir.getAbsolutePath()); + + org.apache.beam.sdk.Pipeline p = org.apache.beam.sdk.Pipeline.create(); + IllegalArgumentException expandEx = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + p.apply( + DeltaIO.readChanges() + .from(tableDir.getAbsolutePath()) + .withStartVersion(0L) + .withEndVersion(1L))); + Assert.assertTrue( + expandEx + .getMessage() + .contains( + "the table schema at version 0 is not compatible with the table schema at the end " + + "version 1")); + + writePipeline + .apply(Create.of(tableDir.getAbsolutePath())) + .apply(ParDo.of(new CreateCDCReadTasksDoFn(null, 0L, null, 1L, null))); + org.apache.beam.sdk.Pipeline.PipelineExecutionException doFnEx = + Assert.assertThrows( + org.apache.beam.sdk.Pipeline.PipelineExecutionException.class, + () -> writePipeline.run().waitUntilFinish()); + Assert.assertTrue( + doFnEx + .getCause() + .getMessage() + .contains( + "the table schema at version 0 is not compatible with the table schema at the end " + + "version 1")); + } + + @Test + public void testReadChangesNonAdditiveSchemaChangeBetweenTimestampsThrows() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-schema-evolution-timestamps"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // Version 0 (committed at timestamp 100000000000L) creates the table with schema (name, age) + // and version 1 (committed at timestamp 200000000000L) drops the column 'age'. + DeltaWriteTestUtils.setupDroppedColumnTable(engine, tableDir.getAbsolutePath()); + + String startTimestamp = java.time.Instant.ofEpochMilli(100000000000L).toString(); + String endTimestamp = java.time.Instant.ofEpochMilli(250000000000L).toString(); + + org.apache.beam.sdk.Pipeline p = org.apache.beam.sdk.Pipeline.create(); + IllegalArgumentException expandEx = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + p.apply( + DeltaIO.readChanges() + .from(tableDir.getAbsolutePath()) + .withStartTimestamp(startTimestamp) + .withEndTimestamp(endTimestamp))); + Assert.assertTrue( + expandEx + .getMessage() + .contains( + "the table schema at version 0 is not compatible with the table schema at the end " + + "version 1")); + } + + @Test + public void testReadChangesColumnAddedAndDroppedWithinRangeThrows() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-column-added-and-dropped"); + String tablePath = tableDir.getAbsolutePath(); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // Version 0 creates the table with schema (name), version 1 adds a column 'age', and version 2 + // drops the column 'age' again. Hence, the schemas at the start and end versions match. + DeltaWriteTestUtils.writeSchemaCommit( + engine, tablePath, 0L, 100000000000L, new StructType().add("name", StringType.STRING)); + StructType initialSchema = + Table.forPath(engine, tablePath).getLatestSnapshot(engine).getSchema(); + DeltaWriteTestUtils.writeSchemaCommit( + engine, tablePath, 1L, 200000000000L, initialSchema.add("age", IntegerType.INTEGER, true)); + DeltaWriteTestUtils.writeSchemaCommit(engine, tablePath, 2L, 300000000000L, initialSchema); + + // Expanding the transform succeeds since it only verifies the schemas at the start and end + // versions. Planning the read tasks fails since it verifies all schema changes within the + // range, and data written with the schema at version 1 can't be read using the schema at the + // end version, which doesn't contain the column 'age'. + readPipeline.apply( + DeltaIO.readChanges().from(tablePath).withStartVersion(0L).withEndVersion(2L)); + org.apache.beam.sdk.Pipeline.PipelineExecutionException doFnEx = + Assert.assertThrows( + org.apache.beam.sdk.Pipeline.PipelineExecutionException.class, + () -> readPipeline.run().waitUntilFinish()); + Assert.assertTrue( + doFnEx + .getCause() + .getMessage() + .contains( + "the table schema at version 1 is not compatible with the table schema at the end " + + "version 2")); + } + + @Test + public void testReadChangesAddedNullableColumnBetweenVersions() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-added-nullable-column"); + String tablePath = tableDir.getAbsolutePath(); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // Version 0 creates the table with schema (name) and inserts two rows. + Schema tableSchema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + DeltaWriteTestUtils.writeAppendCommit( + engine, + tablePath, + 0L, + 100000000000L, + new StructType().add("name", StringType.STRING), + java.util.Arrays.asList( + Row.withSchema(tableSchema).addValues("row-1").build(), + Row.withSchema(tableSchema).addValues("row-2").build())); + + // Version 1 enables column mapping, which the Delta Lake Kernel API requires for schema + // evolution. This is a metadata-only schema change. + DeltaWriteTestUtils.writeTablePropertiesCommit( + engine, + tablePath, + 1L, + 200000000000L, + Collections.singletonMap("delta.columnMapping.mode", "name")); + + // Version 2 adds a nullable column 'age', which is an additive schema change. + StructType latestSchema = + Table.forPath(engine, tablePath).getLatestSnapshot(engine).getSchema(); + DeltaWriteTestUtils.writeSchemaCommit( + engine, tablePath, 2L, 300000000000L, latestSchema.add("age", IntegerType.INTEGER, true)); + + // All changes are read using the schema at the end version, so the rows that were inserted + // before the column 'age' was added get null values for it. + PCollection output = + readPipeline.apply( + DeltaIO.readChanges().from(tablePath).withStartVersion(0L).withEndVersion(2L)); + Assert.assertEquals( + Schema.FieldType.INT32.withNullable(true), output.getSchema().getField("age").getType()); + + PCollection formattedOutput = + output.apply("Format ValueKind, Name, and Age", ParDo.of(new FormatValueKindNameAndAge())); + PAssert.that(formattedOutput).containsInAnyOrder("INSERT:row-1:null", "INSERT:row-2:null"); + + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testReadChangesMetadataOnlySchemaChangeBetweenVersions() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-changes-metadata-only-schema-change"); + String tablePath = tableDir.getAbsolutePath(); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + // Version 0 creates the table with schema (name). + DeltaWriteTestUtils.writeSchemaCommit( + engine, tablePath, 0L, 100000000000L, new StructType().add("name", StringType.STRING)); + + // Version 1 only adds a comment to the existing column, which is a metadata-only schema change. + // The existing field metadata of the column (column mapping ID and physical name) is preserved. + Table table = Table.forPath(engine, tablePath); + StructField nameField = table.getLatestSnapshot(engine).getSchema().get("name"); + FieldMetadata metadataWithComment = + FieldMetadata.builder() + .fromMetadata(nameField.getMetadata()) + .putString("comment", "The name.") + .build(); + DeltaWriteTestUtils.writeSchemaCommit( + engine, + tablePath, + 1L, + 200000000000L, + new StructType().add("name", StringType.STRING, metadataWithComment)); + + // Sanity check that the Delta schemas of the two versions differ (in their field metadata). + Assert.assertNotEquals( + table.getSnapshotAsOfVersion(engine, 0L).getSchema(), + table.getSnapshotAsOfVersion(engine, 1L).getSchema()); + + // Reading changes across the metadata-only schema change succeeds, both when expanding the + // transform and when planning read tasks in CreateCDCReadTasksDoFn. The output is empty since + // the table does not contain any data. + PCollection output = + readPipeline.apply( + DeltaIO.readChanges().from(tablePath).withStartVersion(0L).withEndVersion(1L)); + PAssert.that(output).empty(); + + readPipeline.run().waitUntilFinish(); + } + @Test public void testS3SchemeRegistrationWithAnonymousCredentials() { org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); @@ -1820,6 +2017,15 @@ public void process( } } + private static final class FormatValueKindNameAndAge extends DoFn { + @ProcessElement + public void process( + @Element Row row, ValueKind valueKind, OutputReceiver outputReceiver) { + outputReceiver.output( + valueKind.name() + ":" + row.getString("name") + ":" + row.getInt32("age")); + } + } + private static final class FormatRowWithMetadata extends DoFn { @ProcessElement public void process(@Element Row row, OutputReceiver out) { diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtilsTest.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtilsTest.java new file mode 100644 index 000000000000..01b28ea311f5 --- /dev/null +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaSchemaUtilsTest.java @@ -0,0 +1,220 @@ +/* + * 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.delta; + +import io.delta.kernel.internal.util.ColumnMapping; +import io.delta.kernel.types.ArrayType; +import io.delta.kernel.types.FieldMetadata; +import io.delta.kernel.types.IntegerType; +import io.delta.kernel.types.LongType; +import io.delta.kernel.types.MapType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructField; +import io.delta.kernel.types.StructType; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link DeltaSchemaUtils}. */ +@RunWith(JUnit4.class) +public class DeltaSchemaUtilsTest { + + private static final StructField NAME = + new StructField("name", StringType.STRING, true, columnMappingMetadata(1L, "col-1")); + private static final StructField AGE = + new StructField("age", IntegerType.INTEGER, true, columnMappingMetadata(2L, "col-2")); + private static final StructType SCHEMA = new StructType().add(NAME).add(AGE); + + @Test + public void testIsReadCompatibleIgnoresMetadataOnlyChanges() { + Assert.assertTrue(DeltaSchemaUtils.isReadCompatible(SCHEMA, SCHEMA)); + + StructType schema = + new StructType() + .add("name", StringType.STRING) + .add("nested", new StructType().add("value", LongType.LONG)); + + // Field metadata that does not affect the data, such as column comments, is ignored. This also + // applies to nested fields. + FieldMetadata comment = FieldMetadata.builder().putString("comment", "A comment.").build(); + StructType schemaWithComments = + new StructType() + .add("name", StringType.STRING, comment) + .add("nested", new StructType().add("value", LongType.LONG, comment), comment); + Assert.assertTrue(DeltaSchemaUtils.isReadCompatible(schema, schemaWithComments)); + + // Enabling column mapping on an existing table assigns column IDs to the existing columns and + // sets their physical names to their logical names, so data is still read from the same + // columns. + StructType schemaWithColumnMapping = + new StructType() + .add("name", StringType.STRING, columnMappingMetadata(1L, "name")) + .add( + "nested", + new StructType().add("value", LongType.LONG, columnMappingMetadata(3L, "value")), + columnMappingMetadata(2L, "nested")); + Assert.assertTrue(DeltaSchemaUtils.isReadCompatible(schema, schemaWithColumnMapping)); + } + + @Test + public void testIsReadCompatibleAllowsAdditiveChanges() { + StructField city = + new StructField("city", StringType.STRING, true, columnMappingMetadata(3L, "col-3")); + + // Added nullable columns, at the end or in between existing columns. + Assert.assertTrue(DeltaSchemaUtils.isReadCompatible(SCHEMA, SCHEMA.add(city))); + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible(SCHEMA, new StructType().add(NAME).add(city).add(AGE))); + // Reordered columns, since columns are matched by name. + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible(SCHEMA, new StructType().add(AGE).add(NAME))); + // Relaxed nullability. + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("name", StringType.STRING, false), + new StructType().add("name", StringType.STRING, true))); + + // Added nullable nested field. + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("nested", new StructType().add("value", LongType.LONG)), + new StructType() + .add( + "nested", + new StructType() + .add("value", LongType.LONG) + .add("other", StringType.STRING, true)))); + // Relaxed nullability of array elements and map values. + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("array", new ArrayType(StringType.STRING, false)), + new StructType().add("array", new ArrayType(StringType.STRING, true)))); + Assert.assertTrue( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("map", new MapType(StringType.STRING, LongType.LONG, false)), + new StructType().add("map", new MapType(StringType.STRING, LongType.LONG, true)))); + } + + @Test + public void testIsReadCompatibleRejectsNonAdditiveChanges() { + // Dropped column. + Assert.assertFalse(DeltaSchemaUtils.isReadCompatible(SCHEMA, new StructType().add(NAME))); + // Renamed column. Renames preserve the column mapping metadata of the column. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, + new StructType().add("full_name", StringType.STRING, NAME.getMetadata()).add(AGE))); + // Renamed column where only the case of the name changed, since names are case-sensitive. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, new StructType().add("Name", StringType.STRING, NAME.getMetadata()).add(AGE))); + // Changed data type. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, new StructType().add(NAME).add("age", LongType.LONG, AGE.getMetadata()))); + // Tightened nullability. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, + new StructType().add(NAME).add("age", IntegerType.INTEGER, false, AGE.getMetadata()))); + // Added non-nullable column. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, + SCHEMA.add("city", StringType.STRING, false, columnMappingMetadata(3L, "col-3")))); + + // A column that was dropped and re-added with the same name and type gets a new column ID and + // physical name. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, + new StructType() + .add(NAME) + .add("age", IntegerType.INTEGER, columnMappingMetadata(3L, "col-3")))); + // A re-added column is detected through its new column ID even if it reuses the physical name. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + SCHEMA, + new StructType() + .add(NAME) + .add("age", IntegerType.INTEGER, columnMappingMetadata(3L, "col-2")))); + // A column that was dropped and re-added after enabling column mapping is detected through its + // physical name, which differs from the logical name that was used to read the column before. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("age", IntegerType.INTEGER), + new StructType().add("age", IntegerType.INTEGER, columnMappingMetadata(2L, "col-2")))); + + // Non-additive changes of nested types. + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType() + .add( + "nested", + new StructType().add("value", LongType.LONG).add("other", StringType.STRING)), + new StructType().add("nested", new StructType().add("value", LongType.LONG)))); + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("nested", new StructType().add("value", IntegerType.INTEGER)), + new StructType().add("nested", new StructType().add("value", LongType.LONG)))); + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("array", new ArrayType(IntegerType.INTEGER, true)), + new StructType().add("array", new ArrayType(LongType.LONG, true)))); + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("array", new ArrayType(StringType.STRING, true)), + new StructType().add("array", new ArrayType(StringType.STRING, false)))); + Assert.assertFalse( + DeltaSchemaUtils.isReadCompatible( + new StructType().add("map", new MapType(StringType.STRING, LongType.LONG, true)), + new StructType().add("map", new MapType(StringType.STRING, LongType.LONG, false)))); + } + + @Test + public void testCheckReadCompatible() { + // Does not throw for additive changes. + DeltaSchemaUtils.checkReadCompatible( + SCHEMA, + 1L, + SCHEMA.add("city", StringType.STRING, true, columnMappingMetadata(3L, "col-3")), + 0L, + 2L); + + IllegalArgumentException e = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + DeltaSchemaUtils.checkReadCompatible( + SCHEMA, 1L, new StructType().add(NAME), 0L, 2L)); + Assert.assertTrue( + e.getMessage() + .startsWith( + "Cannot read changes between versions 0 and 2 since the table schema at version 1 " + + "is not compatible with the table schema at the end version 2")); + } + + /** Returns field metadata containing the given column mapping ID and physical name. */ + private static FieldMetadata columnMappingMetadata(long columnId, String physicalName) { + return FieldMetadata.builder() + .putLong(ColumnMapping.COLUMN_MAPPING_ID_KEY, columnId) + .putString(ColumnMapping.COLUMN_MAPPING_PHYSICAL_NAME_KEY, physicalName) + .build(); + } +} diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java index 5ba483d56ce0..536418be0e79 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java @@ -32,6 +32,7 @@ import io.delta.kernel.types.BooleanType; import io.delta.kernel.types.DataType; import io.delta.kernel.types.DateType; +import io.delta.kernel.types.IntegerType; import io.delta.kernel.types.LongType; import io.delta.kernel.types.MapType; import io.delta.kernel.types.StringType; @@ -417,6 +418,101 @@ public static void writeCdcCommit( } } + /** + * Writes a metadata-only Delta commit (i.e. a commit without any data files) that sets the schema + * of the table. + * + *

If {@code expectedVersion} is 0, a new table is created with the given schema, with Change + * Data Feed and column mapping (name mode) enabled. Otherwise, the schema of the existing table + * is evolved to the given schema. + * + *

Note on why this is metadata-only: The Delta Lake Kernel API only supports schema evolution + * for tables that have column mapping enabled, but it does not yet support writing data into such + * tables. + * + *

When evolving the schema, the given schema must preserve the field metadata (column mapping + * IDs and physical names) of existing columns, for example, by deriving the new schema from the + * schema of the latest snapshot. Fields without such metadata are considered to be new columns. + * + * @param engine the Delta Lake {@link Engine} instance to use + * @param tablePath the path of the Delta table to write to + * @param expectedVersion the expected version of the commit to be created + * @param timestamp the timestamp of the commit file + * @param deltaSchema the new schema of the Delta table + * @throws Exception if any error occurs during commit + */ + public static void writeSchemaCommit( + Engine engine, String tablePath, long expectedVersion, long timestamp, StructType deltaSchema) + throws Exception { + + Table table = Table.forPath(engine, tablePath); + TransactionBuilder txnBuilder; + if (expectedVersion == 0) { + Map tableProperties = new HashMap<>(); + tableProperties.put("delta.enableChangeDataFeed", "true"); + tableProperties.put("delta.columnMapping.mode", "name"); + txnBuilder = + table + .createTransactionBuilder(engine, "DeltaTestUtils", Operation.CREATE_TABLE) + .withTableProperties(engine, tableProperties); + } else { + txnBuilder = + table.createTransactionBuilder(engine, "DeltaTestUtils", Operation.MANUAL_UPDATE); + } + Transaction txn = txnBuilder.withSchema(engine, deltaSchema).build(engine); + + TransactionCommitResult result = txn.commit(engine, CloseableIterable.emptyIterable()); + org.junit.Assert.assertEquals(expectedVersion, result.getVersion()); + setCommitFileTimestamp(tablePath, expectedVersion, timestamp); + } + + /** + * Writes a metadata-only Delta commit (i.e. a commit without any data files) that sets the given + * table properties of an existing table. + * + *

For example, this can be used to enable column mapping on an existing table by setting + * {@code delta.columnMapping.mode} to {@code name}. + * + * @param engine the Delta Lake {@link Engine} instance to use + * @param tablePath the path of the Delta table to write to + * @param expectedVersion the expected version of the commit to be created + * @param timestamp the timestamp of the commit file + * @param tableProperties the table properties to set + * @throws Exception if any error occurs during commit + */ + public static void writeTablePropertiesCommit( + Engine engine, + String tablePath, + long expectedVersion, + long timestamp, + Map tableProperties) + throws Exception { + + Transaction txn = + Table.forPath(engine, tablePath) + .createTransactionBuilder(engine, "DeltaTestUtils", Operation.MANUAL_UPDATE) + .withTableProperties(engine, tableProperties) + .build(engine); + + TransactionCommitResult result = txn.commit(engine, CloseableIterable.emptyIterable()); + org.junit.Assert.assertEquals(expectedVersion, result.getVersion()); + setCommitFileTimestamp(tablePath, expectedVersion, timestamp); + } + + /** + * Sets the modification time of the commit file of the given version, which determines the + * timestamp of the commit, to the given timestamp. This is only supported for local tables. + */ + private static void setCommitFileTimestamp(String tablePath, long version, long timestamp) { + if (!tablePath.startsWith("gs://") + && !tablePath.startsWith("s3://") + && !tablePath.startsWith("hdfs://")) { + File commitFile = + new File(new File(tablePath, "_delta_log"), String.format("%020d.json", version)); + commitFile.setLastModified(timestamp); + } + } + /** * Sets up a Delta table with two commit versions containing test rows. * @@ -445,4 +541,33 @@ static List setupTwoVersionTable(Engine engine, String tablePath) throws Ex return java.util.Arrays.asList(row1, row2, row3); } + + /** + * Sets up a Delta table with two metadata-only commit versions where a column is dropped, which + * is a non-additive schema change. + * + *

Version 0 is committed at timestamp 100000000000L and creates the table with schema {@code + * (name STRING, age INTEGER)}. Version 1 is committed at timestamp 200000000000L and drops the + * {@code age} column, resulting in schema {@code (name STRING)}. + * + * @param engine the Delta Lake {@link Engine} instance to use + * @param tablePath the path of the Delta table to create + * @throws Exception if any error occurs during commit + */ + static void setupDroppedColumnTable(Engine engine, String tablePath) throws Exception { + // Commit version 0 + writeSchemaCommit( + engine, + tablePath, + 0L, + 100000000000L, + new StructType().add("name", StringType.STRING).add("age", IntegerType.INTEGER)); + + // Commit version 1. The new schema keeps the existing 'name' field of the latest schema of the + // table to preserve its column mapping metadata. + StructType latestSchema = + Table.forPath(engine, tablePath).getLatestSnapshot(engine).getSchema(); + writeSchemaCommit( + engine, tablePath, 1L, 200000000000L, new StructType().add(latestSchema.get("name"))); + } }