diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java index 524dd4962c89..a203745df28b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java +++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java @@ -311,6 +311,7 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) { if (tableRollback != null) { rollback = new CommitRollback(tableRollback); } + List commitCallbacks = createCommitCallbacks(commitUser, table); return new FileStoreCommitImpl( snapshotCommit, fileIO, @@ -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); } @@ -389,11 +390,17 @@ public InternalRowPartitionComputer partitionComputer() { options.legacyPartitionName()); } - private List createCommitPreCallbacks(FileStoreTable table) { + private List createCommitPreCallbacks( + FileStoreTable table, List commitCallbacks) { List 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; } @@ -591,7 +598,7 @@ public List 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; } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index 4b6776d3396b..86df0fa3424c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java @@ -20,6 +20,7 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; @@ -49,12 +50,13 @@ import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.manifest.ManifestCommittable; import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.SimpleFileEntry; import org.apache.paimon.options.Options; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.SchemaManager; -import org.apache.paimon.schema.TableSchema; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.sink.CommitCallback; +import org.apache.paimon.table.sink.CommitPreCallback; import org.apache.paimon.table.sink.TagCallback; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.DeletionFile; @@ -65,9 +67,13 @@ import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.LocalZonedTimestampType; import org.apache.paimon.types.MapType; import org.apache.paimon.types.MultisetType; import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; +import org.apache.paimon.types.VariantType; +import org.apache.paimon.types.VectorType; import org.apache.paimon.utils.DataFilePathFactories; import org.apache.paimon.utils.FileStorePathFactory; import org.apache.paimon.utils.ManifestReadThreadPool; @@ -75,6 +81,8 @@ import org.apache.paimon.utils.Preconditions; import org.apache.paimon.utils.SnapshotManager; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -99,7 +107,6 @@ import java.util.UUID; import java.util.function.Function; import java.util.stream.Collectors; -import java.util.stream.IntStream; import java.util.stream.Stream; import static org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX; @@ -108,7 +115,7 @@ * A {@link CommitCallback} to create Iceberg compatible metadata, so Iceberg readers can read * Paimon's {@link RawFile}. */ -public class IcebergCommitCallback implements CommitCallback, TagCallback { +public class IcebergCommitCallback implements CommitCallback, CommitPreCallback, TagCallback { private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitCallback.class); @@ -142,6 +149,9 @@ public class IcebergCommitCallback implements CommitCallback, TagCallback { private final FileStorePathFactory fileStorePathFactory; private final IcebergManifestFile manifestFile; private final IcebergManifestList manifestList; + // lazily built to read historical lists written under a legacy manifest version, see + // readManifestListWithFallback + private IcebergManifestList legacyManifestList; private final int formatVersion; private final IndexFileHandler indexFileHandler; @@ -183,28 +193,178 @@ public IcebergCommitCallback(FileStoreTable table, String commitUser) { || formatVersion == IcebergMetadata.FORMAT_VERSION_V3, "Unsupported iceberg format version! Only version 2 or version 3 is valid, but current version is ", formatVersion); + // schema and manifest-legacy config are checked in preflightSchemas, not here, so abort() + // (which emits no Iceberg metadata) is not blocked this.indexFileHandler = table.store().newIndexFileHandler(); this.needAddDvToIceberg = needAddDvToIceberg(); } + /** Creates an instance for tag-only use ({@link #notifyCreation} / {@link #notifyDeletion}). */ + public static IcebergCommitCallback forTagCallbacks(FileStoreTable table) { + return new IcebergCommitCallback(table, ""); + } + + static void checkFormatVersionSupportsSchema(int formatVersion, RowType rowType) { + Collection nanosTimestamps = new LinkedHashSet<>(); + Collection v3OnlyTypes = new LinkedHashSet<>(); + for (DataField field : rowType.getFields()) { + collectRestrictedTypes(field.name(), field.type(), nanosTimestamps, v3OnlyTypes); + } + throwOnNanosecondTimestamps(nanosTimestamps); + if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3) { + return; + } + throwOnV3OnlyTypes(v3OnlyTypes, formatVersion); + } + + static void checkFormatVersionSupportsSchemas(int formatVersion, List schemas) { + Collection nanosTimestamps = new LinkedHashSet<>(); + Collection v3OnlyTypes = new LinkedHashSet<>(); + for (IcebergSchema schema : schemas) { + for (IcebergDataField field : schema.fields()) { + collectRestrictedTypes( + field.name(), field.dataType(), nanosTimestamps, v3OnlyTypes); + } + } + throwOnNanosecondTimestamps(nanosTimestamps); + if (formatVersion >= IcebergMetadata.FORMAT_VERSION_V3) { + return; + } + throwOnV3OnlyTypes(v3OnlyTypes, formatVersion); + } + + private static void throwOnNanosecondTimestamps(Collection nanosTimestamps) { + // Rejected on every format version, unlike the v3-only types below: Paimon writes a + // nanosecond timestamp as Parquet INT96, which Iceberg reads as a microsecond zoned + // timestamp rather than its INT64 timestamp_ns, so the emitted metadata is unreadable. + Preconditions.checkArgument( + nanosTimestamps.isEmpty(), + "Nanosecond-precision timestamps %s are not supported by Iceberg compatibility " + + "because Paimon writes them as Parquet INT96, which Iceberg cannot read " + + "as timestamp_ns. Use a timestamp precision of 6 or less.", + nanosTimestamps); + } + + private static void throwOnV3OnlyTypes(Collection v3OnlyTypes, int formatVersion) { + Preconditions.checkArgument( + v3OnlyTypes.isEmpty(), + "Data types %s require Iceberg format version 3, but the current version is %s. " + + "Please set '%s' = '3'.", + v3OnlyTypes, + formatVersion, + IcebergOptions.FORMAT_VERSION.key()); + } + + private static void collectRestrictedTypes( + String path, + DataType type, + Collection nanosTimestamps, + Collection v3OnlyTypes) { + switch (type.getTypeRoot()) { + case VARIANT: + v3OnlyTypes.add(path + ": " + type.asSQLString()); + break; + case TIMESTAMP_WITHOUT_TIME_ZONE: + if (((TimestampType) type).getPrecision() + >= IcebergDataField.MIN_NANOS_TIMESTAMP_PRECISION) { + nanosTimestamps.add(path + ": " + type.asSQLString()); + } + break; + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + if (((LocalZonedTimestampType) type).getPrecision() + >= IcebergDataField.MIN_NANOS_TIMESTAMP_PRECISION) { + nanosTimestamps.add(path + ": " + type.asSQLString()); + } + break; + case ARRAY: + collectRestrictedTypes( + path + ".element", + ((ArrayType) type).getElementType(), + nanosTimestamps, + v3OnlyTypes); + break; + case MULTISET: + collectRestrictedTypes( + path + ".element", + ((MultisetType) type).getElementType(), + nanosTimestamps, + v3OnlyTypes); + break; + case MAP: + collectRestrictedTypes( + path + ".key", ((MapType) type).getKeyType(), nanosTimestamps, v3OnlyTypes); + collectRestrictedTypes( + path + ".value", + ((MapType) type).getValueType(), + nanosTimestamps, + v3OnlyTypes); + break; + case ROW: + for (DataField field : ((RowType) type).getFields()) { + collectRestrictedTypes( + path + "." + field.name(), field.type(), nanosTimestamps, v3OnlyTypes); + } + break; + case VECTOR: + collectRestrictedTypes( + path + ".element", + ((VectorType) type).getElementType(), + nanosTimestamps, + v3OnlyTypes); + break; + case BOOLEAN: + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + case CHAR: + case VARCHAR: + case BINARY: + case VARBINARY: + case DATE: + case TIME_WITHOUT_TIME_ZONE: + case BLOB: + // exist in format version 2 (BLOB is rejected later, by type conversion) + break; + default: + // fail closed: an unclassified type root must be examined here, or a + // future restricted type could reach Iceberg metadata silently + throw new IllegalArgumentException( + "Type root " + + type.getTypeRoot() + + " has not been classified in the restricted type check: " + + type); + } + } + public static Path catalogTableMetadataPath(FileStoreTable table) { Path icebergDBPath = catalogDatabasePath(table); return new Path(icebergDBPath, String.format("%s/metadata", table.location().getName())); } public static Path catalogDatabasePath(FileStoreTable table) { - Path dbPath = table.location().getParent(); - final String dbSuffix = ".db"; + return catalogDatabasePath(table, resolveStorageLocation(table)); + } + private static IcebergOptions.StorageLocation resolveStorageLocation(FileStoreTable table) { IcebergOptions.StorageType storageType = table.coreOptions().toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); + return table.coreOptions() + .toConfiguration() + .getOptional(IcebergOptions.METADATA_ICEBERG_STORAGE_LOCATION) + .orElse(inferDefaultMetadataLocation(storageType)); + } - IcebergOptions.StorageLocation storageLocation = - table.coreOptions() - .toConfiguration() - .getOptional(IcebergOptions.METADATA_ICEBERG_STORAGE_LOCATION) - .orElse(inferDefaultMetadataLocation(storageType)); + private static Path catalogDatabasePath( + FileStoreTable table, IcebergOptions.StorageLocation storageLocation) { + Path dbPath = table.location().getParent(); + final String dbSuffix = ".db"; + IcebergOptions.StorageType storageType = + table.coreOptions().toConfiguration().get(IcebergOptions.METADATA_ICEBERG_STORAGE); switch (storageLocation) { case TABLE_LOCATION: @@ -262,6 +422,57 @@ public void call(Context context) { context.indexFiles); } + @Override + public void call( + List baseFiles, + List deltaFiles, + List indexFiles, + Snapshot snapshot) { + preflightSchemas(snapshot); + } + + /** + * Validates every schema and partition spec this commit could emit, before any Iceberg metadata + * is written. Throwing here aborts the commit before the Paimon snapshot is published (unlike + * the post-commit {@link #call(Context)}), and {@link #retry(ManifestCommittable)} runs it too + * so a replay cannot slip an unsupported historical schema past the check and orphan the + * manifests it wrote before {@code SchemaCache} rejects it. + */ + private void preflightSchemas(Snapshot snapshot) { + try { + Preconditions.checkArgument( + formatVersion < IcebergMetadata.FORMAT_VERSION_V3 + || !table.coreOptions() + .toConfiguration() + .get(IcebergOptions.MANIFEST_LEGACY_VERSION), + "'%s' cannot be used with Iceberg format version 3: the legacy manifest " + + "schema cannot carry the first_row_id field required by v3 row lineage.", + IcebergOptions.MANIFEST_LEGACY_VERSION.key()); + Path baseMetadataPath = pathFactory.toMetadataPath(snapshot.id() - 1); + boolean hasUsableBase = false; + if (table.fileIO().exists(baseMetadataPath)) { + IcebergMetadata baseMetadata = + IcebergMetadata.fromPath(table.fileIO(), baseMetadataPath); + if (isSameFormatVersion(baseMetadata.formatVersion())) { + hasUsableBase = true; + } + } + // only the current schema is gated: every path that emits older schemas skips + // the never-publishable ones, so a since-remedied schema cannot brick commits + SchemaCache schemaCache = new SchemaCache(); + long latestSchemaId = schemaCache.getLatestSchemaId(); + schemaCache.getValidated(latestSchemaId); + if (!hasUsableBase) { + // without a usable base the partition spec is derived from scratch, and a + // VARIANT partition key cannot be represented there + checkNoVariantPartitionKeys( + table.schema().partitionKeys(), schemaCache.get(latestSchemaId)); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + @Override public void retry(ManifestCommittable committable) { SnapshotManager snapshotManager = table.snapshotManager(); @@ -280,6 +491,7 @@ public void retry(ManifestCommittable committable) { + committable.identifier() + ". This is unexpected.")); long snapshotId = snapshot.id(); + preflightSchemas(snapshot); createMetadata( snapshot, (removedFiles, addedFiles) -> @@ -431,9 +643,25 @@ private void createMetadata( } } catch (IOException e) { throw new UncheckedIOException(e); + } catch (Exception e) { + throw new RuntimeException(e); } } + private static boolean transientReadFailure(Exception e) { + for (Throwable t = e; t != null; t = t.getCause()) { + if (t instanceof FileNotFoundException || t instanceof JsonProcessingException) { + return false; + } + } + for (Throwable t = e; t != null; t = t.getCause()) { + if (t instanceof IOException) { + return true; + } + } + return false; + } + // ------------------------------------------------------------------------------------- // Create metadata afresh // ------------------------------------------------------------------------------------- @@ -515,10 +743,33 @@ private void createMetadataWithoutBase( // current schema follows the latest; the snapshot entry records its own schema int schemaId = (int) schemaCache.getLatestSchemaId(); int snapshotSchemaId = (int) paimonSnapshot.schemaId(); - IcebergSchema icebergSchema = schemaCache.get(schemaId); + IcebergSchema icebergSchema = schemaCache.getValidated(schemaId); List partitionFields = getPartitionFields(table.schema().partitionKeys(), icebergSchema); + List allSchemas = new ArrayList<>(); + for (int id = 0; id <= schemaId; id++) { + if (id == schemaId) { + allSchemas.add(icebergSchema); + continue; + } + try { + allSchemas.add(schemaCache.getValidated(id)); + } catch (RuntimeException neverPublishable) { + if (transientReadFailure(neverPublishable)) { + // a flaky read is not evidence; dropping a valid schema here would + // corrupt time travel for its snapshots permanently + throw neverPublishable; + } + // a historical schema that cannot exist at this format version was never + // mirrored; leaving it out must not fail every rebuild forever + } + } + if (allSchemas.stream().noneMatch(s -> s.schemaId() == (int) paimonSnapshot.schemaId())) { + // provenance must not dangle when the snapshot's own schema was left out + snapshotSchemaId = schemaId; + } + IcebergSnapshotSummary snapshotSummary = computeSnapshotSummary( IcebergSnapshotSummary.APPEND.operation(), paimonSnapshot, metrics); @@ -548,10 +799,6 @@ private void createMetadataWithoutBase( // and external catalogs keep refreshing the same table String tableUuid = inheritUuid != null ? inheritUuid : UUID.randomUUID().toString(); - List allSchemas = - IntStream.rangeClosed(0, schemaId) - .mapToObj(schemaCache::get) - .collect(Collectors.toList()); IcebergMetadata metadata = new IcebergMetadata( formatVersion, @@ -589,12 +836,19 @@ private void createMetadataWithoutBase( table.fileIO().deleteQuietly(metadataPath); written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); } - if (!written && !metadataMatchesSnapshot(snapshotId, paimonSnapshot)) { - // no twin published this snapshot's metadata; fail so the commit retries - throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath); + if (!written) { + if (!metadataMatchesSnapshot(snapshotId, paimonSnapshot)) { + // no twin published this snapshot's metadata; fail so the commit retries + throw new IllegalStateException( + "Failed to replace Iceberg metadata " + metadataPath); + } + // a concurrent callback published this version first; later callbacks derive row + // ids from what is on disk, so adopt the winner wholesale instead of handing a + // divergent twin to the external catalog, and leave clean-up to the winner + metadata = IcebergMetadata.fromPath(table.fileIO(), metadataPath); } - // a delayed callback may still write its metadata (a newer commit extends it), but - // only the current head may move the hint and the external catalog + // a regeneration publishes only as the live head (a rollback legitimately moves the + // hint back); a stale or delayed callback must not move hint or catalog anywhere Long latestAtPublish = table.snapshotManager().latestSnapshotId(); if (latestAtPublish != null && latestAtPublish == snapshotId) { table.fileIO() @@ -602,9 +856,11 @@ private void createMetadataWithoutBase( new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), String.valueOf(snapshotId)); commitToExternalCatalog(metadata, metadataPath, null, null); - // cleanup only after the catalog serves the new head: a skipped or failed - // publication must not delete files an external pointer still references - expireAllBefore(snapshotId); + // cleanup only after the catalog serves the new head, and only by the writer: a + // skipped or failed publication must not delete files a pointer still references + if (written) { + expireAllBefore(snapshotId); + } } } @@ -680,6 +936,8 @@ private void dataSplitToManifestEntries( private List getPartitionFields( List partitionKeys, IcebergSchema icebergSchema) { + checkNoVariantPartitionKeys(partitionKeys, icebergSchema); + Map fields = new HashMap<>(); for (IcebergDataField field : icebergSchema.fields()) { fields.put(field.name(), field); @@ -694,47 +952,21 @@ private List getPartitionFields( return result; } - /** VARIANT needs Iceberg row lineage, which Paimon Iceberg compatibility cannot publish. */ - static void checkVariantNotPublishable(RowType rowType) { - Collection variantFields = new LinkedHashSet<>(); - for (DataField field : rowType.getFields()) { - collectVariantFields(field.name(), field.type(), variantFields); + // Iceberg's identity transform (the only transform Paimon partition values use) rejects + // VARIANT outright, so a VARIANT partition key can never be represented in Iceberg metadata + static void checkNoVariantPartitionKeys( + List partitionKeys, IcebergSchema icebergSchema) { + Set variantPartitionKeys = new LinkedHashSet<>(); + for (IcebergDataField field : icebergSchema.fields()) { + if (partitionKeys.contains(field.name()) && field.dataType() instanceof VariantType) { + variantPartitionKeys.add(field.name()); + } } Preconditions.checkArgument( - variantFields.isEmpty(), - "Columns %s use the VARIANT type, which Paimon Iceberg compatibility cannot " - + "publish: it is an Iceberg format-version-3 type that requires row " - + "lineage.", - variantFields); - } - - private static void collectVariantFields( - String path, DataType type, Collection variantFields) { - switch (type.getTypeRoot()) { - case VARIANT: - variantFields.add(path + ": " + type.asSQLString()); - break; - case ARRAY: - collectVariantFields( - path + ".element", ((ArrayType) type).getElementType(), variantFields); - break; - case MULTISET: - collectVariantFields( - path + ".element", ((MultisetType) type).getElementType(), variantFields); - break; - case MAP: - collectVariantFields(path + ".key", ((MapType) type).getKeyType(), variantFields); - collectVariantFields( - path + ".value", ((MapType) type).getValueType(), variantFields); - break; - case ROW: - for (DataField field : ((RowType) type).getFields()) { - collectVariantFields(path + "." + field.name(), field.type(), variantFields); - } - break; - default: - break; - } + variantPartitionKeys.isEmpty(), + "Partition keys %s have type VARIANT, which Iceberg does not support as a " + + "partition key.", + variantPartitionKeys); } // ------------------------------------------------------------------------------------- @@ -924,11 +1156,13 @@ private void createMetadataWithBase( return; } - // decide the schema story before any manifest is written + // decide the schema story before any manifest is written; the current schema is + // re-gated here because retry() skips the preflight and rejecting later would + // orphan the manifests written meanwhile SchemaCache schemaCache = new SchemaCache(); int schemaId = (int) schemaCache.getLatestSchemaId(); int snapshotSchemaId = (int) snapshot.schemaId(); - IcebergSchema icebergSchema = schemaCache.get(schemaId); + IcebergSchema icebergSchema = schemaCache.getValidated(schemaId); // re-verified each commit: a rollback re-evolution can redefine an already // verified id while this callback only ever sees increasing snapshot ids for (IcebergSchema known : baseMetadata.schemas()) { @@ -968,7 +1202,8 @@ private void createMetadataWithBase( } List baseManifestFileMetas = - manifestList.read(baseMetadata.currentSnapshot().manifestList()); + manifestList.read( + new Path(baseMetadata.currentSnapshot().manifestList()).getName()); // base manifest file for data files List baseDataManifestFileMetas = @@ -998,6 +1233,7 @@ private void createMetadataWithBase( // duplicate. List newDataManifestFileMetas; String operation; + if (isAddOnly) { // Fast case. We don't need to remove files from `baseMetadata`. We only need to append // new metadata files. @@ -1021,23 +1257,22 @@ private void createMetadataWithBase( List newDVManifestFileMetas = new ArrayList<>(); if (needAddDvToIceberg) { if (!indexFiles.isEmpty()) { - // reconstruct the dv index + // the deletion-vector set changed this snapshot: rebuild from the snapshot's live + // state, which is empty when every dv was removed (so a removed dv is not + // re-applied from the base manifests) newDVManifestFileMetas.addAll(createDvManifestFileMetas(snapshot)); } else { - // no new dv index, reuse the old one + // unchanged: keep the base delete manifests newDVManifestFileMetas.addAll(baseDVManifestFileMetas); } } - // compact data manifest file if needed newDataManifestFileMetas = compactMetadataIfNeeded(newDataManifestFileMetas, snapshotId); - String manifestListFileName = - manifestList.writeWithoutRolling( - Stream.concat( - newDataManifestFileMetas.stream(), - newDVManifestFileMetas.stream()) - .collect(Collectors.toList())); + List manifestsToWrite = + Stream.concat(newDataManifestFileMetas.stream(), newDVManifestFileMetas.stream()) + .collect(Collectors.toList()); + String manifestListFileName = manifestList.writeWithoutRolling(manifestsToWrite); SummaryMetrics metrics = new SummaryMetrics(); metrics.addedDataFiles = addedFiles.size(); @@ -1098,15 +1333,40 @@ private void createMetadataWithBase( // append only ids the list does not already carry Set knownSchemaIds = schemas.stream().map(IcebergSchema::schemaId).collect(Collectors.toSet()); - schemas = new ArrayList<>(schemas); - schemas.addAll( - IntStream.rangeClosed(baseMetadata.currentSchemaId() + 1, schemaId) - .filter(id -> !knownSchemaIds.contains(id)) - .mapToObj(schemaCache::get) - .collect(Collectors.toList())); + List added = new ArrayList<>(); + for (int id = baseMetadata.currentSchemaId() + 1; id <= schemaId; id++) { + if (knownSchemaIds.contains(id)) { + continue; + } + if (id == schemaId) { + added.add(icebergSchema); + continue; + } + try { + added.add(schemaCache.getValidated(id)); + } catch (RuntimeException neverPublishable) { + if (transientReadFailure(neverPublishable)) { + // a flaky read is not evidence; dropping a valid schema here would + // corrupt time travel for its snapshots permanently + throw neverPublishable; + } + // a pending schema that was vetoed at its introduction and since remedied + // was never mirrored; leaving it out must not brick every later commit + } + } + if (!added.isEmpty()) { + schemas = new ArrayList<>(schemas); + schemas.addAll(added); + } } // a schema-pointer rollback (validated above): only the current pointer moves + int provenanceSchemaId = snapshotSchemaId; + if (schemas.stream().noneMatch(s -> s.schemaId() == provenanceSchemaId)) { + // provenance must not dangle when the snapshot's own schema was left out + snapshotSchemaId = schemaId; + } + List snapshots = new ArrayList<>(baseMetadata.snapshots()); snapshots.add( new IcebergSnapshot( @@ -1176,12 +1436,21 @@ private void createMetadataWithBase( table.fileIO().deleteQuietly(metadataPath); written = table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson()); } + if (!written && metadataMatchesSnapshot(snapshotId, snapshot)) { + // a concurrent callback published this version first; later callbacks derive row + // ids from what is on disk, so adopt the winner wholesale instead of handing a + // divergent twin to the external catalog, and leave expiry to the winner + metadata = IcebergMetadata.fromPath(table.fileIO(), metadataPath); + } if (!written && !metadataMatchesSnapshot(snapshotId, snapshot)) { // no twin published this snapshot's metadata; fail so the commit retries throw new IllegalStateException("Failed to replace Iceberg metadata " + metadataPath); } - // a delayed callback may still write its metadata (a newer commit extends it), but - // only the current head may move the hint and the external catalog + // publish when advancing the hint (a mirror catch-up moves it forward step by step) + // or when this snapshot is the live head (a rollback legitimately moves it back); + // a delayed replay behind both must not move the hint or the catalog backwards + // publish only as the live head: a rollback legitimately moves the hint back, while a + // stale or delayed callback must not move the hint or the catalog anywhere Long latestAtPublish = table.snapshotManager().latestSnapshotId(); if (latestAtPublish != null && latestAtPublish == snapshotId) { table.fileIO() @@ -1189,13 +1458,15 @@ private void createMetadataWithBase( new Path(pathFactory.metadataDirectory(), VERSION_HINT_FILENAME), String.valueOf(snapshotId)); commitToExternalCatalog(metadata, metadataPath, baseMetadata, baseMetadataPath); - // cleanup only after the catalog serves the new head: a skipped or failed - // publication must not delete files an external pointer still references - deleteApplicableMetadataFiles(snapshotId); - for (int i = 0; i + 1 < toExpireExceptLast.size(); i++) { - expireManifestList( - new Path(toExpireExceptLast.get(i).manifestList()).getName(), - new Path(toExpireExceptLast.get(i + 1).manifestList()).getName()); + // cleanup only after the catalog serves the new head, and only by the writer: a + // skipped or failed publication must not delete files a pointer still references + if (written) { + deleteApplicableMetadataFiles(snapshotId); + for (int i = 0; i + 1 < toExpireExceptLast.size(); i++) { + expireManifestList( + new Path(toExpireExceptLast.get(i).manifestList()).getName(), + new Path(toExpireExceptLast.get(i + 1).manifestList()).getName()); + } } } } @@ -1490,6 +1761,7 @@ private boolean shouldExpire(IcebergSnapshot snapshot, long currentSnapshotId) { - options.get(CoreOptions.SNAPSHOT_TIME_RETAINED).toMillis(); } + @VisibleForTesting private void expireManifestList(String toExpire, String next) { Set metaInUse = new HashSet<>(manifestList.read(next)); for (IcebergManifestFileMeta meta : manifestList.read(toExpire)) { @@ -1942,13 +2214,15 @@ private class SchemaCache { private IcebergSchema get(long schemaId) { return schemas.computeIfAbsent( - schemaId, - id -> { - TableSchema schema = schemaManager.schema(id); - // backstop: reject variant on each schema as it is emitted - checkVariantNotPublishable(schema.logicalRowType()); - return IcebergSchema.create(schema); - }); + schemaId, id -> IcebergSchema.create(schemaManager.schema(id))); + } + + private IcebergSchema getValidated(long schemaId) { + // only schemas a commit newly publishes are gated; history already published + // must not brick later commits (e.g. after the offending column was dropped) + checkFormatVersionSupportsSchema( + formatVersion, schemaManager.schema(schemaId).logicalRowType()); + return get(schemaId); } private long getLatestSchemaId() { diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java index 9862ff7f90c4..5c6e0ef092c5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java @@ -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"; @@ -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: diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 88d6083dec0d..db4a1e83a1b9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -1298,9 +1298,21 @@ CommitResult tryCommitOnce( boolean success; final List finalBaseFiles = baseDataFiles; final List 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) { @@ -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( diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java index a24b5c4c6e9b..bbc021cdc308 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitCleaner.java @@ -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; @@ -50,21 +52,19 @@ public void cleanUpReuseTmpManifests( Pair 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 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( diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCommitCallbackTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCommitCallbackTest.java index e1ecb772db65..a12b4fda609d 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCommitCallbackTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCommitCallbackTest.java @@ -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; @@ -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; @@ -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( diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java index d8a22aaee494..2c909c04fc28 100644 --- a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java @@ -38,6 +38,7 @@ import org.apache.paimon.iceberg.manifest.IcebergManifestFile; import org.apache.paimon.iceberg.manifest.IcebergManifestFileMeta; import org.apache.paimon.iceberg.manifest.IcebergManifestList; +import org.apache.paimon.iceberg.metadata.IcebergDataField; import org.apache.paimon.iceberg.metadata.IcebergMetadata; import org.apache.paimon.iceberg.metadata.IcebergRef; import org.apache.paimon.iceberg.metadata.IcebergSchema; @@ -80,6 +81,7 @@ import org.junit.jupiter.api.io.TempDir; import java.io.File; +import java.io.IOException; import java.math.BigDecimal; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; @@ -102,6 +104,7 @@ import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for Iceberg compatibility. */ @@ -914,6 +917,466 @@ public void testSchemaChange() throws Exception { commit.close(); } + private static long countDeleteManifests( + FileStoreTable table, IcebergPathFactory pathFactory, long snapshotId) + throws IOException { + IcebergMetadata metadata = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(snapshotId)); + IcebergManifestList manifestList = IcebergManifestList.create(table, pathFactory); + return manifestList.read(new Path(metadata.currentSnapshot().manifestList()).getName()) + .stream() + .filter(m -> m.content() == IcebergManifestFileMeta.Content.DELETES) + .count(); + } + + @Test + public void testRollbackToOlderSchemaMirrorsWithoutThrowing() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "3")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + table.createTag("before-evolution", 1); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("w", DataTypes.INT())); + table = table.copyWithLatestSchema(); + write = table.newWrite(commitUser); + commit = table.newCommit(commitUser); + write.write(GenericRow.of(2, 20, 200)); + commit.commit(2, write.prepareCommit(false, 2)); + int evolvedSchemaId = (int) table.snapshotManager().snapshot(2).schemaId(); + write.close(); + commit.close(); + + // rolling back to the pre-evolution snapshot mirrors a snapshot whose schema is older than + // the base metadata's current schema; the post-commit callback must not throw the + // forward-only schema check, which would leave the published rollback with stale metadata + TableCommitImpl rollbackCommit = table.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(table.tagManager().getOrThrow("before-evolution")); + rollbackCommit.close(); + long rolledBackId = table.snapshotManager().latestSnapshotId(); + + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergMetadata rebuilt = + IcebergMetadata.fromPath(table.fileIO(), pathFactory.toMetadataPath(rolledBackId)); + // the table's current schema keeps following the manager's latest, while the + // rolled-back snapshot's entry records the older schema it was written with + assertThat(rebuilt.currentSchemaId()).isEqualTo(evolvedSchemaId); + assertThat(rebuilt.currentSnapshot().schemaId()).isLessThan(evolvedSchemaId); + assertThat(rebuilt.schemas()).anyMatch(s -> s.schemaId() == evolvedSchemaId); + } + + @Test + public void testLegacyManifestVersionRejectedOnV3() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map legacyOptions = new HashMap<>(); + legacyOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + legacyOptions.put(IcebergOptions.MANIFEST_LEGACY_VERSION.key(), "true"); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + legacyOptions); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + assertThatThrownBy(() -> commit.commit(1, write.prepareCommit(false, 1))) + .hasStackTraceContaining("manifest-legacy-version"); + write.close(); + commit.close(); + } + + @Test + public void testLegacyManifestVersionV3AllowsAbort() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map legacyOptions = new HashMap<>(); + legacyOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + legacyOptions.put(IcebergOptions.MANIFEST_LEGACY_VERSION.key(), "true"); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.emptyList(), + -1, + legacyOptions); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + List messages = write.prepareCommit(false, 1); + commit.abort(messages); + write.close(); + commit.close(); + } + + @Test + public void testDroppedVariantSchemaDoesNotBlockV2Commits() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "parquet")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("payload", DataTypes.VARIANT())); + schemaManager.commitChanges(SchemaChange.dropColumn("payload")); + table = table.copyWithLatestSchema(); + + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(2, 20)); + commit2.commit(2, write2.prepareCommit(false, 2)); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(2L); + write2.close(); + commit2.close(); + + IcebergMetadata metadata = + IcebergMetadata.fromPath( + table.fileIO(), new Path(table.location(), "metadata/v2.metadata.json")); + for (IcebergSchema schema : metadata.schemas()) { + for (IcebergDataField field : schema.fields()) { + assertThat(String.valueOf(field.type())).doesNotContain("variant"); + } + } + } + + @Test + public void testLegacyInvalidBaseSchemaRebuiltOnNextCommit() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, Collections.emptyList(), Collections.singletonList("k"), 1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + + FileIO fileIO = table.fileIO(); + Path metadataPath = new Path(table.location(), "metadata/v1.metadata.json"); + String json = fileIO.readFileUtf8(metadataPath); + String doctored = json.replace("\"type\" : \"int\"", "\"type\" : \"variant\""); + if (doctored.equals(json)) { + doctored = json.replace("\"type\":\"int\"", "\"type\":\"variant\""); + } + assertThat(doctored).isNotEqualTo(json); + fileIO.deleteQuietly(metadataPath); + fileIO.overwriteFileUtf8(metadataPath, doctored); + + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(false, 2)); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(2L); + write.close(); + commit.close(); + + IcebergMetadata rebuilt = + IcebergMetadata.fromPath( + fileIO, new Path(table.location(), "metadata/v2.metadata.json")); + assertThat(rebuilt.currentSnapshotId()).isEqualTo(2L); + for (IcebergSchema schema : rebuilt.schemas()) { + for (IcebergDataField field : schema.fields()) { + assertThat(String.valueOf(field.type())).doesNotContain("variant"); + } + } + } + + @Test + public void testRetryValidatesHistoricalSchemaBeforeWriting() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "parquet")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.write(GenericRow.of(2, 20)); + List messages2 = write.prepareCommit(false, 2); + commit.commit(2, messages2); + write.close(); + commit.close(); + + // add a v3-only column after the snapshots: the base metadata (v1) does not carry it, but + // the schema history emitted for snapshot 2 does. retry() must run the same preflight as a + // normal commit and reject before writing (and orphaning) any Iceberg manifest. + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("payload", DataTypes.VARIANT())); + table = table.copyWithLatestSchema(); + + IcebergPathFactory pf = new IcebergPathFactory(new Path(table.location(), "metadata")); + table.fileIO().deleteQuietly(pf.toMetadataPath(2)); + int before = table.fileIO().listStatus(pf.metadataDirectory()).length; + + TableCommitImpl commit2 = table.newCommit(commitUser); + Map> retryMessages = new HashMap<>(); + retryMessages.put(2L, messages2); + assertThatThrownBy(() -> commit2.filterAndCommit(retryMessages)) + .hasStackTraceContaining("require Iceberg format version 3"); + assertThat(table.fileIO().listStatus(pf.metadataDirectory()).length).isEqualTo(before); + commit2.close(); + } + + @Test + public void testRejectedCommitDoesNotLeaveOrphanedManifests() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + Map legacyOptions = new HashMap<>(); + legacyOptions.put(IcebergOptions.MANIFEST_LEGACY_VERSION.key(), "true"); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + legacyOptions); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + FileIO fileIO = table.fileIO(); + Path manifestDir = new Path(table.location(), "manifest"); + int manifestFileCountBefore = fileIO.listStatus(manifestDir).length; + + FileStoreTable upgraded = + table.copy(Collections.singletonMap(IcebergOptions.FORMAT_VERSION.key(), "3")); + TableWriteImpl write2 = upgraded.newWrite(commitUser); + TableCommitImpl commit2 = upgraded.newCommit(commitUser); + write2.write(GenericRow.of(2, 20)); + assertThatThrownBy(() -> commit2.commit(2, write2.prepareCommit(false, 2))) + .hasStackTraceContaining("cannot be used with Iceberg format version 3"); + assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(1L); + + int manifestFileCountAfter = fileIO.listStatus(manifestDir).length; + assertThat(manifestFileCountAfter).isEqualTo(manifestFileCountBefore); + + write2.close(); + commit2.close(); + } + + @Test + public void testVariantPartitionKeyRejectedOnAnalyzeCommit() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, + new String[] {"k", "payload"}); + Map options = new HashMap<>(); + options.put(CoreOptions.FILE_FORMAT.key(), "parquet"); + options.put(IcebergOptions.FORMAT_VERSION.key(), "3"); + assertThatThrownBy( + () -> + createPaimonTable( + rowType, + Collections.singletonList("payload"), + Collections.emptyList(), + -1, + options)) + .hasStackTraceContaining("VariantType") + .hasStackTraceContaining("partition"); + } + + @Test + public void testFormatVersionUpgradeRecoversFromLegacyInvalidBase() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, Collections.emptyList(), Collections.singletonList("k"), 1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + + FileIO fileIO = table.fileIO(); + Path metadataPath = new Path(table.location(), "metadata/v1.metadata.json"); + String json = fileIO.readFileUtf8(metadataPath); + String doctored = json.replace("\"type\" : \"int\"", "\"type\" : \"variant\""); + if (doctored.equals(json)) { + doctored = json.replace("\"type\":\"int\"", "\"type\":\"variant\""); + } + assertThat(doctored).isNotEqualTo(json); + fileIO.deleteQuietly(metadataPath); + fileIO.overwriteFileUtf8(metadataPath, doctored); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges( + SchemaChange.setOption(IcebergOptions.FORMAT_VERSION.key(), "3")); + table = table.copy(table.schemaManager().latest().get()); + write.close(); + commit.close(); + + TableWriteImpl write2 = table.newWrite(commitUser); + TableCommitImpl commit2 = table.newCommit(commitUser); + write2.write(GenericRow.of(2, 20)); + commit2.commit(2, write2.prepareCommit(false, 2)); + write2.close(); + commit2.close(); + + IcebergMetadata metadata = + IcebergMetadata.fromPath( + fileIO, new Path(table.location(), "metadata/v2.metadata.json")); + assertThat(metadata.formatVersion()).isEqualTo(IcebergMetadata.FORMAT_VERSION_V3); + } + + @Test + public void testRollbackToAsLatestSkipsDroppedIntermediateVariantSchema() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "parquet")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + table.createTag("target", 1); + + write.write(GenericRow.of(2, 20)); + commit.commit(2, write.prepareCommit(false, 2)); + write.close(); + commit.close(); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("payload", DataTypes.VARIANT())); + schemaManager.commitChanges(SchemaChange.dropColumn("payload")); + FileStoreTable tableWithEvolvedSchema = table.copyWithLatestSchema(); + + TableCommitImpl rollbackCommit = tableWithEvolvedSchema.newCommit(commitUser); + rollbackCommit.rollbackToAsLatest(tableWithEvolvedSchema.tagManager().getOrThrow("target")); + assertThat(tableWithEvolvedSchema.snapshotManager().latestSnapshotId()).isEqualTo(3L); + rollbackCommit.close(); + + IcebergMetadata metadata = + IcebergMetadata.fromPath( + table.fileIO(), new Path(table.location(), "metadata/v3.metadata.json")); + assertThat(metadata.currentSnapshotId()).isEqualTo(3L); + for (IcebergSchema schema : metadata.schemas()) { + for (IcebergDataField field : schema.fields()) { + assertThat(String.valueOf(field.type())).doesNotContain("variant"); + } + } + } + + @Test + public void testDeleteTagSucceedsAfterUnsafeCurrentSchemaChange() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "parquet")); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + write.write(GenericRow.of(1, 10)); + commit.commit(1, write.prepareCommit(false, 1)); + write.close(); + commit.close(); + + table.createTag("t1", 1); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("payload", DataTypes.VARIANT())); + FileStoreTable tableWithUnsafeSchema = table.copyWithLatestSchema(); + + assertThatCode(() -> tableWithUnsafeSchema.deleteTag("t1")).doesNotThrowAnyException(); + } + + @Test + public void testAbortSucceedsWithUnsafeCurrentSchema() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.INT()}, new String[] {"k", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, + Collections.emptyList(), + Collections.singletonList("k"), + 1, + Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "parquet")); + + SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location()); + schemaManager.commitChanges(SchemaChange.addColumn("payload", DataTypes.VARIANT())); + FileStoreTable tableWithUnsafeSchema = table.copyWithLatestSchema(); + + String commitUser = UUID.randomUUID().toString(); + assertThatCode( + () -> { + TableCommitImpl commit = tableWithUnsafeSchema.newCommit(commitUser); + commit.abort(Collections.emptyList()); + commit.close(); + }) + .doesNotThrowAnyException(); + } + @Test public void testIcebergSnapshotExpire() throws Exception { RowType rowType = @@ -2560,12 +3023,17 @@ private FileStoreTable createPaimonTable( options.set(CoreOptions.BUCKET, numBuckets); options.set( IcebergOptions.METADATA_ICEBERG_STORAGE, IcebergOptions.StorageType.TABLE_LOCATION); - options.set(CoreOptions.FILE_FORMAT, "avro"); + if (!customOptions.containsKey(CoreOptions.FILE_FORMAT.key())) { + options.set(CoreOptions.FILE_FORMAT, "avro"); + } options.set(CoreOptions.TARGET_FILE_SIZE, MemorySize.ofKibiBytes(32)); - options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 4); - options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 8); + if (!customOptions.containsKey(IcebergOptions.COMPACT_MIN_FILE_NUM.key())) { + options.set(IcebergOptions.COMPACT_MIN_FILE_NUM, 8); + } options.set(IcebergOptions.METADATA_DELETE_AFTER_COMMIT, true); - options.set(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX, 1); + if (!customOptions.containsKey(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX.key())) { + options.set(IcebergOptions.METADATA_PREVIOUS_VERSIONS_MAX, 1); + } options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE, MemorySize.ofKibiBytes(8)); Schema schema = new Schema(rowType.getFields(), partitionKeys, primaryKeys, options.toMap(), ""); @@ -2641,17 +3109,4 @@ private void parseAvroFields( } } } - - @Test - public void testVariantIsNotPublishableToIceberg() { - RowType withVariant = - RowType.of( - new DataType[] {DataTypes.INT(), DataTypes.VARIANT()}, - new String[] {"k", "payload"}); - assertThatThrownBy(() -> IcebergCommitCallback.checkVariantNotPublishable(withVariant)) - .hasMessageContaining("VARIANT type"); - - IcebergCommitCallback.checkVariantNotPublishable( - RowType.of(new DataType[] {DataTypes.INT()}, new String[] {"k"})); - } } diff --git a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java index 052f76bda7ca..4db44695fb47 100644 --- a/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java +++ b/paimon-iceberg/src/test/java/org/apache/paimon/iceberg/IcebergRestMetadataCommitterTest.java @@ -63,6 +63,7 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -924,24 +925,12 @@ public void testAbandonedBaseWithSameIdIsNotExtendedInCatalog() throws Exception assertThat(getIcebergResult()) .containsExactlyInAnyOrder("Record(1, 10)", "Record(3, 30)", "Record(4, 40)"); - org.apache.paimon.iceberg.metadata.IcebergMetadata localMetadata = - org.apache.paimon.iceberg.metadata.IcebergMetadata.fromPath( - table.fileIO(), - new org.apache.paimon.fs.Path( - IcebergCommitCallback.catalogTableMetadataPath(table), - "v3.metadata.json")); - String localIdentity = - localMetadata.snapshots().stream() - .filter(snap -> snap.snapshotId() == 2) - .findFirst() - .get() - .summary() - .get("paimon-commit-identity"); + String liveIdentity = table.snapshotManager().snapshot(2).uuid(); Table icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); org.apache.iceberg.Snapshot catalogSnapshot2 = icebergTable.snapshot(2); if (catalogSnapshot2 != null) { assertThat(catalogSnapshot2.summary().get("paimon-commit-identity")) - .isEqualTo(localIdentity); + .isEqualTo(liveIdentity); } } @@ -1051,6 +1040,7 @@ public void testWithIncorrectBase() throws Exception { "Record(6, 60)"); icebergTable = restCatalog.loadTable(TableIdentifier.of("mydb", "t")); assertThat(icebergTable.currentSnapshot().snapshotId()).isEqualTo(7); + // the re-enable gap is mirrored snapshot by snapshot, so the catalog carries them all assertThat(ImmutableList.copyOf(icebergTable.snapshots()).size()).isEqualTo(2); write.write(GenericRow.of(4, 41)); @@ -1223,6 +1213,16 @@ public void testToRestLocationNormalisesScheme() { assertThat(IcebergRestMetadataCommitter.toRestLocation(null)).isNull(); } + private static List readAllRecords(Table icebergTable) throws IOException { + List records = new ArrayList<>(); + try (CloseableIterable result = IcebergGenerics.read(icebergTable).build()) { + for (Record record : result) { + records.add(record.toString()); + } + } + return records; + } + private static class TestRecord { private final BinaryRow partition; private final GenericRow record;