Skip to content

[iceberg] Reject schemas the Iceberg mirror cannot represent before the commit #9199

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
plusplusjiajia wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

[iceberg] Reject schemas the Iceberg mirror cannot represent before the commit #9199

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,7 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
if (tableRollback != null) {
rollback = new CommitRollback(tableRollback);
}
List<CommitCallback> commitCallbacks = createCommitCallbacks(commitUser, table);
return new FileStoreCommitImpl(
snapshotCommit,
fileIO,
Expand All @@ -327,8 +328,8 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
this::newScan,
newStatsFileHandler(),
bucketMode(),
createCommitPreCallbacks(table),
createCommitCallbacks(commitUser, table),
createCommitPreCallbacks(table, commitCallbacks),
commitCallbacks,
conflictDetectFactory,
rollback);
}
Expand Down Expand Up @@ -389,11 +390,17 @@ public InternalRowPartitionComputer partitionComputer() {
options.legacyPartitionName());
}

private List<CommitPreCallback> createCommitPreCallbacks(FileStoreTable table) {
private List<CommitPreCallback> createCommitPreCallbacks(
FileStoreTable table, List<CommitCallback> commitCallbacks) {
List<CommitPreCallback> callbacks = new ArrayList<>();
if (options.isChainTable()) {
callbacks.add(new ChainTableCommitPreCallback(table));
}
// reuse the same Iceberg callback instance: it validates before the commit too
commitCallbacks.stream()
.filter(callback -> callback instanceof IcebergCommitCallback)
.map(callback -> (IcebergCommitCallback) callback)
.forEach(callbacks::add);
return callbacks;
}

Expand Down Expand Up @@ -591,7 +598,7 @@ public List<TagCallback> createTagCallbacks(FileStoreTable table) {
}
if (options.toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE)
!= IcebergOptions.StorageType.DISABLED) {
callbacks.add(new IcebergCommitCallback(table, ""));
callbacks.add(IcebergCommitCallback.forTagCallbacks(table));
}
return callbacks;
}
Expand Down
466 changes: 370 additions & 96 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java
View file
Open in desktop

Large diffs are not rendered by default.

11 changes: 9 additions & 2 deletions paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,9 @@
@JsonIgnoreProperties(ignoreUnknown = true)
public class IcebergDataField {

/** Minimum precision mapped to the v3-only nanosecond types (timestamp[tz]_ns). */
public static final int MIN_NANOS_TIMESTAMP_PRECISION = 7;

private static final String FIELD_ID = "id";
private static final String FIELD_NAME = "name";
private static final String FIELD_REQUIRED = "required";
Expand Down Expand Up @@ -187,13 +190,17 @@ private static Object toTypeObject(DataType dataType, int fieldId, int depth) {
Preconditions.checkArgument(
timestampPrecision >= 3 && timestampPrecision <= 9,
"Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9.");
return timestampPrecision >= 7 ? "timestamp_ns" : "timestamp";
return timestampPrecision >= MIN_NANOS_TIMESTAMP_PRECISION
? "timestamp_ns"
: "timestamp";
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
int timestampLtzPrecision = ((LocalZonedTimestampType) dataType).getPrecision();
Preconditions.checkArgument(
timestampLtzPrecision >= 3 && timestampLtzPrecision <= 9,
"Paimon Iceberg compatibility only support timestamp type with precision from 3 to 9.");
return timestampLtzPrecision >= 7 ? "timestamptz_ns" : "timestamptz";
return timestampLtzPrecision >= MIN_NANOS_TIMESTAMP_PRECISION
? "timestamptz_ns"
: "timestamptz";
case VARIANT:
return "variant";
case ARRAY:
Expand Down
30 changes: 25 additions & 5 deletions paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1298,9 +1298,21 @@ CommitResult tryCommitOnce(
boolean success;
final List<SimpleFileEntry> finalBaseFiles = baseDataFiles;
final List<ManifestEntry> finalDeltaFiles = deltaFiles;
commitPreCallbacks.forEach(
callback ->
callback.call(finalBaseFiles, finalDeltaFiles, indexFiles, newSnapshot));
try {
commitPreCallbacks.forEach(
callback ->
callback.call(
finalBaseFiles, finalDeltaFiles, indexFiles, newSnapshot));
} catch (RuntimeException e) {
// a pre-commit callback vetoed the commit: the manifests prepared above were
// never referenced by a published snapshot and must be cleaned up, exactly
// like a preparation failure
commitCleaner.cleanUpReuseTmpManifests(
deltaManifestList, changelogManifestList, oldIndexManifest, indexManifest);
commitCleaner.cleanUpNoReuseTmpManifests(
baseManifestList, mergeBeforeManifests, mergeAfterManifests);
throw e;
}
try {
success = commitSnapshotImpl(latestSnapshot, newSnapshot, deltaStatistics);
} catch (Exception e) {
Expand Down Expand Up @@ -1533,8 +1545,16 @@ public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
// They may veto the rollback by throwing (e.g. a chain-table snapshot branch rejects a
// pure-DELETE overwrite that would drop a snapshot partition still anchoring delta
// partitions), in which case the rollback snapshot is never created.
commitPreCallbacks.forEach(
callback -> callback.call(baseFiles, deltaFiles, indexChanges, newSnapshot));
try {
commitPreCallbacks.forEach(
callback -> callback.call(baseFiles, deltaFiles, indexChanges, newSnapshot));
} catch (RuntimeException e) {
// the base and delta manifests written above were never referenced by a
// published snapshot and must be cleaned up, exactly like a regular commit
commitCleaner.cleanUpManifestList(baseManifestList);
commitCleaner.cleanUpManifestList(deltaManifestList);
throw e;
}

boolean success =
commitSnapshotImpl(
Expand Down
22 changes: 11 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.utils.Pair;

import javax.annotation.Nullable;

import java.util.List;
import java.util.Objects;
import java.util.Set;
Expand All @@ -50,21 +52,19 @@ public void cleanUpReuseTmpManifests(
Pair<String, Long> changelogManifestList,
String oldIndexManifest,
String newIndexManifest) {
if (deltaManifestList != null) {
for (ManifestFileMeta manifest : manifestList.read(deltaManifestList.getKey())) {
manifestFile.delete(manifest.fileName());
}
manifestList.delete(deltaManifestList.getKey());
}
cleanUpManifestList(deltaManifestList);
cleanUpManifestList(changelogManifestList);
cleanIndexManifest(oldIndexManifest, newIndexManifest);
}

if (changelogManifestList != null) {
for (ManifestFileMeta manifest : manifestList.read(changelogManifestList.getKey())) {
/** Deletes a manifest list and every manifest file it references. */
public void cleanUpManifestList(@Nullable Pair<String, Long> list) {
if (list != null) {
for (ManifestFileMeta manifest : manifestList.read(list.getKey())) {
manifestFile.delete(manifest.fileName());
}
manifestList.delete(changelogManifestList.getKey());
manifestList.delete(list.getKey());
}

cleanIndexManifest(oldIndexManifest, newIndexManifest);
}

public void cleanUpNoReuseTmpManifests(
Expand Down
161 changes: 161 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCommitCallbackTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,15 @@

import org.apache.paimon.CoreOptions;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.metadata.IcebergDataField;
import org.apache.paimon.iceberg.metadata.IcebergMetadata;
import org.apache.paimon.iceberg.metadata.IcebergSchema;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.tag.Tag;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.TagManager;

import org.junit.jupiter.api.BeforeEach;
Expand All @@ -33,6 +39,8 @@
import org.mockito.Answers;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Collections;
import java.util.Optional;
import java.util.stream.Stream;

Expand Down Expand Up @@ -103,6 +111,159 @@ private static void setField(Object target, String fieldName, Object value) thro
field.set(target, value);
}

@Test
void testFormatVersion2RejectsVariantType() {
RowType rowType =
RowType.of(
new DataField(0, "id", DataTypes.INT()),
new DataField(1, "payload", DataTypes.VARIANT()));

assertThatThrownBy(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
IcebergMetadata.FORMAT_VERSION_V2, rowType))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("VARIANT")
.hasMessageContaining("metadata.iceberg.format-version");
}

@Test
void testFormatVersion2RejectsNestedVariantType() {
RowType rowType =
RowType.of(
new DataField(
0,
"nested",
DataTypes.ARRAY(
DataTypes.ROW(
new DataField(
1, "payload", DataTypes.VARIANT())))));

assertThatThrownBy(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
IcebergMetadata.FORMAT_VERSION_V2, rowType))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("nested.element.payload: VARIANT");
}

@Test
void testFormatVersion2RejectsVariantInEmittedSchemas() {
IcebergSchema schema =
new IcebergSchema(
0,
Arrays.asList(
new IcebergDataField(1, "id", true, "int", null),
new IcebergDataField(2, "payload", false, "variant", null)));

assertThatThrownBy(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchemas(
IcebergMetadata.FORMAT_VERSION_V2,
Collections.singletonList(schema)))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("payload: VARIANT")
.hasMessageContaining("metadata.iceberg.format-version");

assertThatCode(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchemas(
IcebergMetadata.FORMAT_VERSION_V3,
Collections.singletonList(schema)))
.doesNotThrowAnyException();
}

@Test
void testVariantPartitionKeyRejected() {
IcebergSchema schema =
new IcebergSchema(
0,
Arrays.asList(
new IcebergDataField(1, "k", true, "int", null),
new IcebergDataField(2, "payload", false, "variant", null)));

assertThatThrownBy(
() ->
IcebergCommitCallback.checkNoVariantPartitionKeys(
Collections.singletonList("payload"), schema))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("payload")
.hasMessageContaining("VARIANT");
}

@Test
void testNonVariantPartitionKeyAllowed() {
IcebergSchema schema =
new IcebergSchema(
0,
Arrays.asList(
new IcebergDataField(1, "k", true, "int", null),
new IcebergDataField(2, "payload", false, "variant", null)));

assertThatCode(
() ->
IcebergCommitCallback.checkNoVariantPartitionKeys(
Collections.singletonList("k"), schema))
.doesNotThrowAnyException();
}

@Test
void testNanosecondTimestampsRejectedOnAllVersions() {
// Paimon writes nanosecond timestamps as Parquet INT96, which Iceberg reads as a
// microsecond zoned timestamp -- not its INT64 timestamp_ns. Emitting timestamp_ns
// metadata over such files is unreadable, so the type is rejected on every version.
for (int formatVersion :
new int[] {IcebergMetadata.FORMAT_VERSION_V2, IcebergMetadata.FORMAT_VERSION_V3}) {
RowType timestampNs = RowType.of(new DataField(0, "ts", DataTypes.TIMESTAMP(9)));
assertThatThrownBy(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
formatVersion, timestampNs))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Nanosecond-precision");

RowType timestampLtzNs =
RowType.of(new DataField(0, "ts", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(7)));
assertThatThrownBy(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
formatVersion, timestampLtzNs))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Nanosecond-precision");
}
}

@Test
void testFormatVersion3AllowsVariantType() {
RowType rowType = RowType.of(new DataField(0, "payload", DataTypes.VARIANT()));

assertThatCode(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
IcebergMetadata.FORMAT_VERSION_V3, rowType))
.doesNotThrowAnyException();
}

@Test
void testFormatVersion2AllowsNonV3Types() {
RowType rowType =
RowType.of(
new DataField(0, "id", DataTypes.INT()),
new DataField(1, "name", DataTypes.STRING()),
new DataField(2, "ts", DataTypes.TIMESTAMP(6)),
new DataField(
3,
"nested",
DataTypes.MAP(
DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT()))));

assertThatCode(
() ->
IcebergCommitCallback.checkFormatVersionSupportsSchema(
IcebergMetadata.FORMAT_VERSION_V2, rowType))
.doesNotThrowAnyException();
}

@ParameterizedTest(name = "StorageType: {0}")
@MethodSource("provideMetadataPathsWithStorageType")
void testCatalogTableMetadataPathWithStorageType(
Expand Down
Loading
Loading