This is an automated email from the ASF dual-hosted git repository.
chamikaramj 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 be3d44e841f Updates Delta CDC read to fail when schemas are different
at... (#40868)
be3d44e841f is described below
commit be3d44e841f41faf166bbea385fc5ebae8ba2e20
Author: Chamikara Jayalath <[email protected]>
AuthorDate: Wed Oct 28 16:47:08 2026 -0700
Updates Delta CDC read to fail when schemas are different at... (#40868)
---
.../beam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +-
.../beam_PreCommit_Java_Delta_IO_Direct.json | 2 +-
.../beam/sdk/io/delta/CreateCDCReadTasksDoFn.java | 44 ++++-
.../java/org/apache/beam/sdk/io/delta/DeltaIO.java | 39 +++-
.../apache/beam/sdk/io/delta/DeltaSchemaUtils.java | 188 ++++++++++++++++++
.../org/apache/beam/sdk/io/delta/DeltaIOTest.java | 206 +++++++++++++++++++
.../beam/sdk/io/delta/DeltaSchemaUtilsTest.java | 220 +++++++++++++++++++++
.../beam/sdk/io/delta/DeltaWriteTestUtils.java | 125 ++++++++++++
8 files changed, 814 insertions(+), 12 deletions(-)
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 9a4fbff0c0b..a7c9525dc7e 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 e3d6056a5de..c537844dc84 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 9a0049a8651..a6dfc01c811 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.DeltaLogActionUtils.DeltaAction;
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 @@ class CreateCDCReadTasksDoFn extends DoFn<String,
DeltaCDCReadTask> {
resolvedStartVersion, resolvedEndVersion));
}
- // 2. Load snapshot at resolvedEndVersion to get the scanStateRow
- // We use endVersion's schema because it represents the latest schema in
the
- // read range
- // which handles schema evolution (older files will just lack new columns).
- Snapshot endSnapshot = table.getSnapshotAsOfVersion(engine,
resolvedEndVersion);
+ // 2. Load snapshots at resolvedStartVersion and resolvedEndVersion. All
changes within the
+ // range
+ // are read using the schema at resolvedEndVersion, which handles schema
evolution (older files
+ // will just lack new columns). This is only supported if the range only
spans additive schema
+ // changes (see DeltaSchemaUtils), which is verified for the schema at
resolvedStartVersion here
+ // and for all schema changes within the range while reading the commits
of the range below.
+ Snapshot startSnapshot = table.getSnapshotAsOfVersion(engine,
resolvedStartVersion);
+ Snapshot endSnapshot =
+ resolvedStartVersion == resolvedEndVersion
+ ? startSnapshot
+ : table.getSnapshotAsOfVersion(engine, resolvedEndVersion);
+ io.delta.kernel.types.StructType startSchema = startSnapshot.getSchema();
+ io.delta.kernel.types.StructType endSchema = endSnapshot.getSchema();
+ if (startSchema == null || endSchema == null) {
+ throw new IllegalStateException("Table schema is null.");
+ }
+ DeltaSchemaUtils.checkReadCompatible(
+ startSchema, resolvedStartVersion, endSchema, resolvedStartVersion,
resolvedEndVersion);
Scan scan = endSnapshot.getScanBuilder().build();
Row scanState = scan.getScanState(engine);
SerializableRow serializableScanState = new SerializableRow(scanState);
- // 3. Load snapshot at resolvedStartVersion to initialize the CommitRange
- Snapshot startSnapshot = table.getSnapshotAsOfVersion(engine,
resolvedStartVersion);
-
CommitRangeBuilder rangeBuilder =
TableManager.loadCommitRange(
tablePath,
CommitRangeBuilder.CommitBoundary.atVersion(resolvedStartVersion));
@@ -129,9 +140,11 @@ class CreateCDCReadTasksDoFn extends DoFn<String,
DeltaCDCReadTask> {
// We need both CDC and ADD actions.
// If a commit version has CDC files, we only read CDC files.
// If a commit version has no CDC files, we read ADD files (inserts).
+ // We also need METADATA actions to verify all schema changes within the
range.
Set<DeltaAction> 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<ColumnarBatch> batchIter =
@@ -144,11 +157,26 @@ class CreateCDCReadTasksDoFn extends DoFn<String,
DeltaCDCReadTask> {
int timestampIdx = batch.getSchema().indexOf("timestamp");
int cdcIdx = batch.getSchema().indexOf("cdc");
int addIdx = batch.getSchema().indexOf("add");
+ int metadataIdx =
batch.getSchema().indexOf(DeltaAction.METADATA.colName);
for (int i = 0; i < batch.getSize(); i++) {
long version = batch.getColumnVector(versionIdx).getLong(i);
long timestamp = batch.getColumnVector(timestampIdx).getLong(i);
+ // A METADATA action means that the table metadata, which includes
the table schema, was
+ // updated in this version. Data written in this or later versions
(until the next
+ // update) uses the new schema, so it also has to be readable using
the end schema.
+ // Verifying every schema change within the range, and not only the
schemas at the start
+ // and end versions, catches non-additive changes that are reverted
within the range, for
+ // example, a column that was added and then dropped again. Since
tasks are only emitted
+ // after all commits of the range were read, no tasks are emitted if
this check fails.
+ if (metadataIdx >= 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 c63fdcacc1c..b745d13160e 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 class DeltaIO {
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 class DeltaIO {
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 00000000000..98259645799
--- /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}.
+ *
+ * <p>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.
+ *
+ * <p>This mirrors the schema compatibility check of change data feed reads
in Delta Lake. The
+ * following changes are allowed:
+ *
+ * <ul>
+ * <li>Adding nullable fields, including nested fields. Records read from
data written with
+ * {@code dataSchema} get null values for these fields.
+ * <li>Relaxing the nullability of a field from non-nullable to nullable.
+ * <li>Reordering fields, since fields are matched by name.
+ * <li>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.
+ * </ul>
+ *
+ * <p>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.
+ *
+ * <p>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 342c4b94761..d27c31d43e1 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.BinaryType;
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 class DeltaIOTest {
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<Row> output =
+ readPipeline.apply(
+
DeltaIO.readChanges().from(tablePath).withStartVersion(0L).withEndVersion(2L));
+ Assert.assertEquals(
+ Schema.FieldType.INT32.withNullable(true),
output.getSchema().getField("age").getType());
+
+ PCollection<String> 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<Row> 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 class DeltaIOTest {
}
}
+ private static final class FormatValueKindNameAndAge extends DoFn<Row,
String> {
+ @ProcessElement
+ public void process(
+ @Element Row row, ValueKind valueKind, OutputReceiver<String>
outputReceiver) {
+ outputReceiver.output(
+ valueKind.name() + ":" + row.getString("name") + ":" +
row.getInt32("age"));
+ }
+ }
+
private static final class FormatRowWithMetadata extends DoFn<Row, String> {
@ProcessElement
public void process(@Element Row row, OutputReceiver<String> 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 00000000000..01b28ea311f
--- /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 5ba483d56ce..536418be0e7 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.internal.data.GenericRow;
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 final class DeltaWriteTestUtils {
}
}
+ /**
+ * Writes a metadata-only Delta commit (i.e. a commit without any data
files) that sets the schema
+ * of the table.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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<String, String> 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.
+ *
+ * <p>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<String, String> 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 @@ public final class DeltaWriteTestUtils {
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.
+ *
+ * <p>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")));
+ }
}
