diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java index 67c75c3a63d1..0db0b538cf91 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java @@ -73,6 +73,8 @@ import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException; import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException; import org.apache.spark.sql.catalyst.analysis.ViewUtil; +import org.apache.spark.sql.connector.catalog.Changelog; +import org.apache.spark.sql.connector.catalog.ChangelogContext; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.NamespaceChange; import org.apache.spark.sql.connector.catalog.StagedTable; @@ -192,6 +194,17 @@ public Table loadTable(Identifier ident, long timestampMicros) throws NoSuchTabl return load(ident, TimeTravel.timestampMicros(timestampMicros)); } + @Override + public Changelog loadChangelog( + Identifier ident, ChangelogContext context, CaseInsensitiveStringMap options) + throws NoSuchTableException { + try { + return new SparkChangelogTable(icebergCatalog.loadTable(buildIdentifier(ident)), context); + } catch (org.apache.iceberg.exceptions.NoSuchTableException e) { + throw new NoSuchTableException(ident); + } + } + @Override public boolean tableExists(Identifier ident) { if (isPathIdentifier(ident)) { diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java index d754f84b276d..375df452d57c 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkSessionCatalog.java @@ -43,6 +43,8 @@ import org.apache.spark.sql.catalyst.analysis.ViewAlreadyExistsException; import org.apache.spark.sql.connector.catalog.CatalogExtension; import org.apache.spark.sql.connector.catalog.CatalogPlugin; +import org.apache.spark.sql.connector.catalog.Changelog; +import org.apache.spark.sql.connector.catalog.ChangelogContext; import org.apache.spark.sql.connector.catalog.FunctionCatalog; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.NamespaceChange; @@ -224,6 +226,17 @@ public Table loadTable(Identifier ident, long timestamp) throws NoSuchTableExcep } } + @Override + public Changelog loadChangelog( + Identifier ident, ChangelogContext context, CaseInsensitiveStringMap options) + throws NoSuchTableException { + try { + return icebergCatalog.loadChangelog(ident, context, options); + } catch (NoSuchTableException e) { + return getSessionCatalog().loadChangelog(ident, context, options); + } + } + @Override public void invalidateTable(Identifier ident) { // We do not need to check whether the table exists and whether diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java index eb8e5e63f430..9fcfb38a75ec 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java @@ -32,6 +32,7 @@ import org.apache.iceberg.DeletedRowsScanTask; import org.apache.iceberg.ScanTaskGroup; import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.CloseableIterator; @@ -42,12 +43,15 @@ import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; import org.apache.spark.sql.catalyst.expressions.JoinedRow; +import org.apache.spark.sql.connector.catalog.Changelog; import org.apache.spark.sql.connector.read.PartitionReader; import org.apache.spark.unsafe.types.UTF8String; class ChangelogRowReader extends BaseRowReader implements PartitionReader { + private final SparkChangelogReadMode readMode; + ChangelogRowReader(SparkInputPartition partition) { this( partition.table(), @@ -69,9 +73,13 @@ class ChangelogRowReader extends BaseRowReader table, fileIO, taskGroup, - ChangelogUtil.dropChangelogMetadata(expectedSchema), + dataSchema(table, expectedSchema), caseSensitive, cacheDeleteFilesOnExecutors); + this.readMode = + expectedSchema.findField(SparkChangelogTable.COMMIT_VERSION) != null + ? SparkChangelogReadMode.SPARK_CDC + : SparkChangelogReadMode.ICEBERG_CHANGELOG; } @Override @@ -86,16 +94,40 @@ protected CloseableIterator open(ChangelogScanTask task) { return cdcRows.iterator(); } - private static InternalRow changelogMetadata(ChangelogScanTask task) { + private InternalRow changelogMetadata(ChangelogScanTask task) { InternalRow metadataRow = new GenericInternalRow(3); - metadataRow.update(0, UTF8String.fromString(task.operation().name())); - metadataRow.update(1, task.changeOrdinal()); - metadataRow.update(2, task.commitSnapshotId()); + if (readMode.isSparkCdc()) { + Snapshot snapshot = table().snapshot(task.commitSnapshotId()); + Preconditions.checkNotNull( + snapshot, "Cannot find snapshot for changelog task: %s", task.commitSnapshotId()); + metadataRow.update(0, UTF8String.fromString(changeType(task))); + metadataRow.update(1, snapshot.sequenceNumber()); + metadataRow.update(2, snapshot.timestampMillis() * 1000); + } else { + metadataRow.update(0, UTF8String.fromString(task.operation().name())); + metadataRow.update(1, task.changeOrdinal()); + metadataRow.update(2, task.commitSnapshotId()); + } return metadataRow; } + private static Schema dataSchema(Table table, Schema expectedSchema) { + return expectedSchema.findField(SparkChangelogTable.COMMIT_VERSION) != null + ? SparkChangelogTable.cdcDataSchema(table) + : ChangelogUtil.dropChangelogMetadata(expectedSchema); + } + + private static String changeType(ChangelogScanTask task) { + return switch (task.operation()) { + case INSERT -> Changelog.CHANGE_TYPE_INSERT; + case DELETE -> Changelog.CHANGE_TYPE_DELETE; + case UPDATE_BEFORE -> Changelog.CHANGE_TYPE_UPDATE_PREIMAGE; + case UPDATE_AFTER -> Changelog.CHANGE_TYPE_UPDATE_POSTIMAGE; + }; + } + private CloseableIterable openChangelogScanTask(ChangelogScanTask task) { if (task instanceof AddedRowsScanTask) { return openAddedRowsScanTask((AddedRowsScanTask) task); diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogMicroBatchStream.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogMicroBatchStream.java new file mode 100644 index 000000000000..d7e011a8bea5 --- /dev/null +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogMicroBatchStream.java @@ -0,0 +1,155 @@ +/* + * 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.iceberg.spark.source; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import org.apache.iceberg.ChangelogScanTask; +import org.apache.iceberg.ChangelogUtil; +import org.apache.iceberg.IncrementalChangelogScan; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.SparkReadConf; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.broadcast.Broadcast; +import org.apache.spark.sql.connector.read.streaming.Offset; +import org.apache.spark.sql.connector.read.streaming.ReadLimit; + +/** + * A minimal changelog stream that advances at Iceberg snapshot boundaries. + * + *

Each planned range contains complete snapshots, ensuring that all rows from a commit remain in + * the same Spark micro-batch. + */ +class SparkChangelogMicroBatchStream extends SparkMicroBatchStreamBase { + + private Broadcast plannedTableBroadcast = null; + + SparkChangelogMicroBatchStream( + JavaSparkContext sparkContext, + Table table, + SparkReadConf readConf, + Schema projection, + String checkpointLocation) { + super( + sparkContext, + table, + table::io, + readConf, + projection, + checkpointLocation, + () -> configuredInitialOffset(readConf)); + } + + private static StreamingOffset configuredInitialOffset(SparkReadConf readConf) { + return readConf.startSnapshotId() != null + ? new StreamingOffset(readConf.startSnapshotId(), 0, false) + : StreamingOffset.START_OFFSET; + } + + @Override + protected StreamingOffset latestStreamingOffset() { + table().refresh(); + Snapshot latest = table().currentSnapshot(); + Long configuredEndSnapshotId = readConf().endSnapshotId(); + if (configuredEndSnapshotId != null) { + latest = table().snapshot(configuredEndSnapshotId); + } + + return latest != null + ? new StreamingOffset(latest.snapshotId(), 0, false) + : StreamingOffset.START_OFFSET; + } + + @Override + public Offset latestOffset(Offset startOffset, ReadLimit limit) { + Preconditions.checkArgument( + startOffset instanceof StreamingOffset, "Invalid start offset: %s", startOffset); + + StreamingOffset latestOffset = (StreamingOffset) latestOffset(); + return latestOffset.equals(StreamingOffset.START_OFFSET) || latestOffset.equals(startOffset) + ? null + : latestOffset; + } + + @Override + protected List> planTaskGroups( + StreamingOffset startOffset, StreamingOffset endOffset) { + if (endOffset.equals(StreamingOffset.START_OFFSET) || startOffset.equals(endOffset)) { + return Lists.newArrayList(); + } + + table().refresh(); + if (!startOffset.equals(StreamingOffset.START_OFFSET)) { + Preconditions.checkState( + table().snapshot(startOffset.snapshotId()) != null, + "Cannot load changelog start offset at expired or removed snapshot: %s", + startOffset.snapshotId()); + } + + Preconditions.checkState( + table().snapshot(endOffset.snapshotId()) != null, + "Cannot load changelog end offset at expired or removed snapshot: %s", + endOffset.snapshotId()); + + IncrementalChangelogScan scan = + table() + .newIncrementalChangelogScan() + .caseSensitive(readConf().caseSensitive()) + .project(ChangelogUtil.changelogSchema(SparkChangelogTable.cdcDataSchema(table()))); + if (!startOffset.equals(StreamingOffset.START_OFFSET)) { + scan = scan.fromSnapshotExclusive(startOffset.snapshotId()); + } + scan = scan.toSnapshot(endOffset.snapshotId()); + + List> taskGroups; + try (CloseableIterable> groups = scan.planTasks()) { + taskGroups = Lists.newArrayList(groups); + } catch (IOException e) { + throw new UncheckedIOException("Failed to close Iceberg changelog task groups", e); + } + + return taskGroups; + } + + @Override + protected Broadcast
tableBroadcast() { + if (plannedTableBroadcast != null) { + plannedTableBroadcast.unpersist(false); + } + + this.plannedTableBroadcast = + sparkContext().broadcast(SerializableTableWithSize.copyOf(table())); + return plannedTableBroadcast; + } + + @Override + protected void stopStream() { + if (plannedTableBroadcast != null) { + plannedTableBroadcast.unpersist(false); + plannedTableBroadcast = null; + } + } +} diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogReadMode.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogReadMode.java new file mode 100644 index 000000000000..b9b43cd0220e --- /dev/null +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogReadMode.java @@ -0,0 +1,33 @@ +/* + * 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.iceberg.spark.source; + +enum SparkChangelogReadMode { + ICEBERG_CHANGELOG, + SPARK_CDC, + EMPTY_SPARK_CDC; + + boolean isSparkCdc() { + return this != ICEBERG_CHANGELOG; + } + + boolean isEmpty() { + return this == EMPTY_SPARK_CDC; + } +} diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java index 57ccf92b9651..edb85582e28d 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java @@ -42,6 +42,7 @@ import org.apache.spark.sql.connector.read.Scan; import org.apache.spark.sql.connector.read.Statistics; import org.apache.spark.sql.connector.read.SupportsReportStatistics; +import org.apache.spark.sql.connector.read.streaming.MicroBatchStream; import org.apache.spark.sql.types.StructType; class SparkChangelogScan implements Scan, SupportsReportStatistics { @@ -56,6 +57,7 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics { private final List filters; private final Long startSnapshotId; private final Long endSnapshotId; + private final SparkChangelogReadMode readMode; // lazy variables private List> taskGroups = null; @@ -68,7 +70,24 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics { SparkReadConf readConf, Schema projection, List filters) { + this( + spark, + table, + scan, + readConf, + projection, + filters, + SparkChangelogReadMode.ICEBERG_CHANGELOG); + } + SparkChangelogScan( + SparkSession spark, + Table table, + IncrementalChangelogScan scan, + SparkReadConf readConf, + Schema projection, + List filters, + SparkChangelogReadMode readMode) { SparkSchemaUtil.validateMetadataColumnReferences(table.schema(), projection); this.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext()); @@ -79,6 +98,7 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics { this.filters = filters != null ? filters : Collections.emptyList(); this.startSnapshotId = readConf.startSnapshotId(); this.endSnapshotId = readConf.endSnapshotId(); + this.readMode = readMode; if (scan == null) { this.taskGroups = Collections.emptyList(); } @@ -113,6 +133,16 @@ public Batch toBatch() { hashCode()); } + @Override + public MicroBatchStream toMicroBatchStream(String checkpointLocation) { + if (!readMode.isSparkCdc()) { + throw new UnsupportedOperationException("Changelog streaming is only supported through CDC"); + } + + return new SparkChangelogMicroBatchStream( + sparkContext, table, readConf, projection, checkpointLocation); + } + private List> taskGroups() { if (taskGroups == null) { try (CloseableIterable> groups = scan.planTasks()) { diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScanBuilder.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScanBuilder.java index 43b8a36507db..b8af8eb31b0f 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScanBuilder.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScanBuilder.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.spark.source; +import org.apache.iceberg.ChangelogUtil; import org.apache.iceberg.IncrementalChangelogScan; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; @@ -26,18 +27,44 @@ import org.apache.iceberg.spark.SparkReadOptions; import org.apache.iceberg.util.SnapshotUtil; import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.connector.expressions.filter.Predicate; import org.apache.spark.sql.connector.read.Scan; import org.apache.spark.sql.connector.read.SupportsPushDownLimit; import org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns; import org.apache.spark.sql.connector.read.SupportsPushDownV2Filters; +import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.util.CaseInsensitiveStringMap; public class SparkChangelogScanBuilder extends BaseSparkScanBuilder implements SupportsPushDownV2Filters, SupportsPushDownRequiredColumns, SupportsPushDownLimit { + private final SparkChangelogReadMode readMode; + SparkChangelogScanBuilder( SparkSession spark, Table table, Schema schema, CaseInsensitiveStringMap options) { + this(spark, table, schema, options, SparkChangelogReadMode.ICEBERG_CHANGELOG); + } + + SparkChangelogScanBuilder( + SparkSession spark, + Table table, + Schema schema, + CaseInsensitiveStringMap options, + SparkChangelogReadMode readMode) { super(spark, table, schema, options); + this.readMode = readMode; + } + + @Override + public void pruneColumns(StructType requestedType) { + if (!readMode.isSparkCdc()) { + super.pruneColumns(requestedType); + } + } + + @Override + public Predicate[] pushPredicates(Predicate[] predicates) { + return readMode.isSparkCdc() ? predicates : super.pushPredicates(predicates); } @Override @@ -65,7 +92,9 @@ public Scan build() { SparkReadOptions.START_TIMESTAMP, SparkReadOptions.END_TIMESTAMP); - if (startTimestamp != null) { + if (readMode.isEmpty()) { + return emptyChangelogScan(); + } else if (startTimestamp != null) { if (noSnapshotsAfter(startTimestamp)) { return emptyChangelogScan(); } @@ -79,9 +108,15 @@ public Scan build() { } } - Schema projection = projectionWithMetadataColumns(); - IncrementalChangelogScan scan = buildIcebergScan(projection, startSnapshotId, endSnapshotId); - return new SparkChangelogScan(spark(), table(), scan, readConf(), projection, filters()); + Schema readProjection = projectionWithMetadataColumns(); + Schema scanProjection = + readMode.isSparkCdc() + ? ChangelogUtil.changelogSchema(SparkChangelogTable.cdcDataSchema(table())) + : readProjection; + IncrementalChangelogScan scan = + buildIcebergScan(scanProjection, startSnapshotId, endSnapshotId); + return new SparkChangelogScan( + spark(), table(), scan, readConf(), readProjection, filters(), readMode); } private IncrementalChangelogScan buildIcebergScan( @@ -114,7 +149,8 @@ private SparkChangelogScan emptyChangelogScan() { null /* no scan */, readConf(), projectionWithMetadataColumns(), - filters()); + filters(), + readMode); } private boolean noSnapshotsAfter(long timestamp) { diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogTable.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogTable.java index bdafca27fbb8..fcd00fe9e46b 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogTable.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogTable.java @@ -18,38 +18,121 @@ */ package org.apache.iceberg.spark.source; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.TreeMap; import org.apache.iceberg.ChangelogUtil; +import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; +import org.apache.iceberg.TableUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkReadOptions; import org.apache.iceberg.spark.SparkSchemaUtil; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.SnapshotUtil; import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.connector.catalog.Changelog; +import org.apache.spark.sql.connector.catalog.ChangelogContext; +import org.apache.spark.sql.connector.catalog.ChangelogRange; +import org.apache.spark.sql.connector.catalog.Column; import org.apache.spark.sql.connector.catalog.MetadataColumn; import org.apache.spark.sql.connector.catalog.SupportsMetadataColumns; import org.apache.spark.sql.connector.catalog.SupportsRead; import org.apache.spark.sql.connector.catalog.TableCapability; +import org.apache.spark.sql.connector.expressions.NamedReference; import org.apache.spark.sql.connector.read.ScanBuilder; import org.apache.spark.sql.types.StructType; import org.apache.spark.sql.util.CaseInsensitiveStringMap; +/** + * Iceberg changelog relation used both as {@code table.changes} and Spark 4.2 {@link Changelog}. + * + *

The Table API keeps Iceberg's changelog columns ({@code _change_type}, {@code + * _change_ordinal}, {@code _commit_snapshot_id}). The Changelog API maps those onto Spark CDC + * columns ({@code _change_type}, {@code _commit_version}, {@code _commit_timestamp}). + */ public class SparkChangelogTable - implements org.apache.spark.sql.connector.catalog.Table, SupportsRead, SupportsMetadataColumns { + implements org.apache.spark.sql.connector.catalog.Table, + SupportsRead, + SupportsMetadataColumns, + Changelog { public static final String TABLE_NAME = "changes"; + static final String COMMIT_VERSION = "_commit_version"; + static final String COMMIT_TIMESTAMP = "_commit_timestamp"; + private static final Set CAPABILITIES = - ImmutableSet.of(TableCapability.BATCH_READ); + ImmutableSet.of(TableCapability.BATCH_READ, TableCapability.MICRO_BATCH_READ); + + private static final Types.NestedField ROW_ID_FIELD = + Types.NestedField.required( + MetadataColumns.ROW_ID.fieldId(), + MetadataColumns.ROW_ID.name(), + MetadataColumns.ROW_ID.type(), + MetadataColumns.ROW_ID.doc()); + private static final Types.NestedField ROW_VERSION_FIELD = + Types.NestedField.required( + MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId(), + MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name(), + MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.type(), + MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.doc()); + + private static final Types.NestedField COMMIT_VERSION_FIELD = + Types.NestedField.required( + Integer.MAX_VALUE - 109, + COMMIT_VERSION, + Types.LongType.get(), + "Iceberg snapshot sequence number"); + private static final Types.NestedField COMMIT_TIMESTAMP_FIELD = + Types.NestedField.required( + Integer.MAX_VALUE - 110, + COMMIT_TIMESTAMP, + Types.TimestampType.withZone(), + "Iceberg snapshot commit timestamp"); private final Table table; - private final Schema schema; + private final Schema icebergChangelogSchema; + private final RangeOptions rangeOptions; + private final Schema sparkCdcSchema; + private final Column[] sparkCdcColumns; private SparkSession lazySpark = null; private StructType lazySparkSchema = null; public SparkChangelogTable(Table table) { + this(table, RangeOptions.icebergChangelog()); + } + + public SparkChangelogTable(Table table, ChangelogContext context) { + this(table, rangeOptions(table, context)); + } + + private SparkChangelogTable(Table table, RangeOptions rangeOptions) { this.table = table; - this.schema = ChangelogUtil.changelogSchema(table.schema()); + this.icebergChangelogSchema = ChangelogUtil.changelogSchema(table.schema()); + this.rangeOptions = rangeOptions; + Preconditions.checkArgument( + !rangeOptions.readMode().isSparkCdc() || TableUtil.supportsRowLineage(table), + "Spark CDC requires an Iceberg table with row lineage"); + this.sparkCdcSchema = + TypeUtil.join( + cdcDataSchema(table), + new Schema(MetadataColumns.CHANGE_TYPE, COMMIT_VERSION_FIELD, COMMIT_TIMESTAMP_FIELD)); + this.sparkCdcColumns = toColumns(sparkCdcSchema); + } + + static Schema cdcDataSchema(Table table) { + return TypeUtil.join(table.schema(), new Schema(ROW_ID_FIELD, ROW_VERSION_FIELD)); } @Override @@ -60,20 +143,66 @@ public String name() { @Override public StructType schema() { if (lazySparkSchema == null) { + Schema schema = + rangeOptions.readMode().isSparkCdc() ? sparkCdcSchema : icebergChangelogSchema; this.lazySparkSchema = SparkSchemaUtil.convert(schema); } return lazySparkSchema; } + @Override + public Column[] columns() { + return rangeOptions.readMode().isSparkCdc() + ? sparkCdcColumns + : toColumns(icebergChangelogSchema); + } + @Override public Set capabilities() { return CAPABILITIES; } + @Override + public boolean containsCarryoverRows() { + return true; + } + + @Override + public boolean containsIntermediateChanges() { + return true; + } + + @Override + public boolean representsUpdateAsDeleteAndInsert() { + return true; + } + + @Override + public NamedReference[] rowId() { + return new NamedReference[] {Spark3Util.toNamedReference(MetadataColumns.ROW_ID.name())}; + } + + @Override + public NamedReference rowVersion() { + return Spark3Util.toNamedReference(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name()); + } + @Override public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { - return new SparkChangelogScanBuilder(spark(), table, schema, options); + if (!rangeOptions.readMode().isSparkCdc()) { + return new SparkChangelogScanBuilder(spark(), table, icebergChangelogSchema, options); + } + + Map scanOptions = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + scanOptions.putAll(options.asCaseSensitiveMap()); + scanOptions.putAll(rangeOptions.options()); + return new SparkChangelogScanBuilder( + spark(), + table, + sparkCdcSchema, + new CaseInsensitiveStringMap(scanOptions), + rangeOptions.readMode()); } private SparkSession spark() { @@ -94,4 +223,149 @@ public MetadataColumn[] metadataColumns() { SparkMetadataColumns.IS_DELETED, }; } + + private static RangeOptions rangeOptions(Table table, ChangelogContext context) { + ChangelogRange range = context.range(); + if (range instanceof ChangelogRange.UnboundedRange) { + return RangeOptions.scan(Collections.emptyMap()); + } else if (range instanceof ChangelogRange.VersionRange) { + return versionRangeOptions(table, (ChangelogRange.VersionRange) range); + } else if (range instanceof ChangelogRange.TimestampRange) { + return timestampRangeOptions(table, (ChangelogRange.TimestampRange) range); + } else { + throw new UnsupportedOperationException("Unsupported Spark changelog range: " + range); + } + } + + private static RangeOptions versionRangeOptions(Table table, ChangelogRange.VersionRange range) { + Snapshot start = snapshotWithSequenceNumber(table, range.startingVersion()); + Snapshot end = + range + .endingVersion() + .map(version -> snapshotWithSequenceNumber(table, version)) + .orElse(table.currentSnapshot()); + if (end == null) { + return RangeOptions.empty(); + } + + Long startExclusive = range.startingBoundInclusive() ? start.parentId() : start.snapshotId(); + Long endInclusive = range.endingBoundInclusive() ? end.snapshotId() : end.parentId(); + return snapshotRangeOptions(table, startExclusive, endInclusive); + } + + private static RangeOptions timestampRangeOptions( + Table table, ChangelogRange.TimestampRange range) { + List snapshots = currentAncestorsInCommitOrder(table); + Snapshot start = + snapshots.stream() + .filter( + snapshot -> + range.startingBoundInclusive() + ? snapshot.timestampMillis() * 1000 >= range.startingTimestamp() + : snapshot.timestampMillis() * 1000 > range.startingTimestamp()) + .findFirst() + .orElse(null); + Snapshot end = + snapshots.stream() + .filter( + snapshot -> + range.endingTimestamp().isEmpty() + || (range.endingBoundInclusive() + ? snapshot.timestampMillis() * 1000 <= range.endingTimestamp().get() + : snapshot.timestampMillis() * 1000 < range.endingTimestamp().get())) + .reduce((left, right) -> right) + .orElse(null); + + if (start == null || end == null || start.sequenceNumber() > end.sequenceNumber()) { + return RangeOptions.empty(); + } + + return snapshotRangeOptions(table, start.parentId(), end.snapshotId()); + } + + private static RangeOptions snapshotRangeOptions( + Table table, Long startExclusive, Long endInclusive) { + if (endInclusive == null) { + return RangeOptions.empty(); + } + + Snapshot end = table.snapshot(endInclusive); + if (end == null) { + return RangeOptions.empty(); + } + + if (startExclusive != null) { + Snapshot start = table.snapshot(startExclusive); + if (start != null && start.sequenceNumber() >= end.sequenceNumber()) { + return RangeOptions.empty(); + } + } + + Map options = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + if (startExclusive != null) { + options.put(SparkReadOptions.START_SNAPSHOT_ID, String.valueOf(startExclusive)); + } + options.put(SparkReadOptions.END_SNAPSHOT_ID, String.valueOf(endInclusive)); + return RangeOptions.scan(options); + } + + private static Snapshot snapshotWithSequenceNumber(Table table, String version) { + long sequenceNumber; + try { + sequenceNumber = Long.parseLong(version); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid Iceberg snapshot sequence number: " + version, e); + } + + return Lists.newArrayList(SnapshotUtil.currentAncestors(table)).stream() + .filter(snapshot -> snapshot.sequenceNumber() == sequenceNumber) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Cannot find Iceberg snapshot with sequence number: " + version)); + } + + private static List currentAncestorsInCommitOrder(Table table) { + List snapshots = Lists.newArrayList(SnapshotUtil.currentAncestors(table)); + Collections.reverse(snapshots); + return snapshots; + } + + private static Column[] toColumns(Schema schema) { + StructType sparkSchema = SparkSchemaUtil.convert(schema); + return Arrays.stream(sparkSchema.fields()) + .map(field -> Column.create(field.name(), field.dataType(), field.nullable())) + .toArray(Column[]::new); + } + + private static class RangeOptions { + private final Map options; + private final SparkChangelogReadMode readMode; + + private RangeOptions(Map options, SparkChangelogReadMode readMode) { + this.options = options; + this.readMode = readMode; + } + + private static RangeOptions icebergChangelog() { + return new RangeOptions(Collections.emptyMap(), SparkChangelogReadMode.ICEBERG_CHANGELOG); + } + + private static RangeOptions scan(Map options) { + return new RangeOptions(options, SparkChangelogReadMode.SPARK_CDC); + } + + private static RangeOptions empty() { + return new RangeOptions(Collections.emptyMap(), SparkChangelogReadMode.EMPTY_SPARK_CDC); + } + + private Map options() { + return options; + } + + private SparkChangelogReadMode readMode() { + return readMode; + } + } } diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java index 7adf3c633cd0..18bcd1ef07db 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java @@ -18,67 +18,36 @@ */ package org.apache.iceberg.spark.source; -import java.io.BufferedWriter; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.OutputStreamWriter; -import java.io.UncheckedIOException; -import java.nio.charset.StandardCharsets; import java.util.List; import java.util.function.Supplier; -import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.CombinedScanTask; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.Schema; -import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; -import org.apache.iceberg.hadoop.HadoopFileIO; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.FileIO; -import org.apache.iceberg.io.InputFile; -import org.apache.iceberg.io.OutputFile; -import org.apache.iceberg.relocated.com.google.common.base.Joiner; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.spark.SparkReadConf; -import org.apache.iceberg.types.Types; import org.apache.iceberg.util.TableScanUtil; import org.apache.spark.api.java.JavaSparkContext; -import org.apache.spark.broadcast.Broadcast; -import org.apache.spark.sql.connector.read.InputPartition; -import org.apache.spark.sql.connector.read.PartitionReaderFactory; -import org.apache.spark.sql.connector.read.streaming.MicroBatchStream; import org.apache.spark.sql.connector.read.streaming.Offset; import org.apache.spark.sql.connector.read.streaming.ReadLimit; -import org.apache.spark.sql.connector.read.streaming.SupportsTriggerAvailableNow; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerAvailableNow { - private static final Joiner SLASH = Joiner.on("/"); +public class SparkMicroBatchStream extends SparkMicroBatchStreamBase { private static final Logger LOG = LoggerFactory.getLogger(SparkMicroBatchStream.class); - private static final Types.StructType EMPTY_GROUPING_KEY_TYPE = Types.StructType.of(); - private final Table table; - private final Supplier fileIO; - private final SparkReadConf readConf; - private final boolean caseSensitive; - private final String projection; - private final Broadcast

tableBroadcast; - private final Broadcast fileIOBroadcast; private final long splitSize; private final int splitLookback; private final long splitOpenFileCost; private final boolean localityPreferred; - private final StreamingOffset initialOffset; private final long fromTimestamp; private final int maxFilesPerMicroBatch; private final int maxRecordsPerMicroBatch; - private final boolean cacheDeleteFilesOnExecutors; private SparkMicroBatchPlanner planner; - private StreamingOffset lastOffsetForTriggerAvailableNow; SparkMicroBatchStream( JavaSparkContext sparkContext, @@ -87,61 +56,52 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerA SparkReadConf readConf, Schema projection, String checkpointLocation) { - this.table = table; - this.fileIO = fileIO; - this.readConf = readConf; - this.caseSensitive = readConf.caseSensitive(); - this.projection = SchemaParser.toJson(projection); + super( + sparkContext, + table, + fileIO, + readConf, + projection, + checkpointLocation, + () -> { + table.refresh(); + return MicroBatchUtils.determineStartingOffset(table, readConf.streamFromTimestamp()); + }); this.localityPreferred = readConf.localityEnabled(); - this.tableBroadcast = sparkContext.broadcast(SerializableTableWithSize.copyOf(table)); - this.fileIOBroadcast = sparkContext.broadcast(SerializableFileIOWithSize.wrap(fileIO.get())); this.splitSize = readConf.splitSize(); this.splitLookback = readConf.splitLookback(); this.splitOpenFileCost = readConf.splitOpenFileCost(); this.fromTimestamp = readConf.streamFromTimestamp(); this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch(); this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch(); - this.cacheDeleteFilesOnExecutors = readConf.cacheDeleteFilesOnExecutors(); - - InitialOffsetStore initialOffsetStore = - new InitialOffsetStore( - table, checkpointLocation, fromTimestamp, sparkContext.hadoopConfiguration()); - this.initialOffset = initialOffsetStore.initialOffset(); } @Override - public Offset latestOffset() { - table.refresh(); - if (table.currentSnapshot() == null) { + protected StreamingOffset latestStreamingOffset() { + table().refresh(); + if (table().currentSnapshot() == null) { return StreamingOffset.START_OFFSET; } - if (table.currentSnapshot().timestampMillis() < fromTimestamp) { + if (table().currentSnapshot().timestampMillis() < fromTimestamp) { return StreamingOffset.START_OFFSET; } - Snapshot latestSnapshot = table.currentSnapshot(); + Snapshot latestSnapshot = table().currentSnapshot(); return new StreamingOffset( - latestSnapshot.snapshotId(), MicroBatchUtils.addedFilesCount(table, latestSnapshot), false); + latestSnapshot.snapshotId(), + MicroBatchUtils.addedFilesCount(table(), latestSnapshot), + false); } @Override - public InputPartition[] planInputPartitions(Offset start, Offset end) { - Preconditions.checkArgument( - end instanceof StreamingOffset, "Invalid end offset: %s is not a StreamingOffset", end); - Preconditions.checkArgument( - start instanceof StreamingOffset, - "Invalid start offset: %s is not a StreamingOffset", - start); - - if (end.equals(StreamingOffset.START_OFFSET)) { - return new InputPartition[0]; + protected List planTaskGroups( + StreamingOffset startOffset, StreamingOffset endOffset) { + if (endOffset.equals(StreamingOffset.START_OFFSET)) { + return Lists.newArrayList(); } - StreamingOffset endOffset = (StreamingOffset) end; - StreamingOffset startOffset = (StreamingOffset) start; - // Initialize planner if not already done (for resume scenarios) if (planner == null) { initializePlanner(startOffset, endOffset); @@ -151,68 +111,30 @@ public InputPartition[] planInputPartitions(Offset start, Offset end) { CloseableIterable splitTasks = TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTasks), splitSize); - List combinedScanTasks = - Lists.newArrayList( - TableScanUtil.planTasks(splitTasks, splitSize, splitLookback, splitOpenFileCost)); - String[][] locations = computePreferredLocations(combinedScanTasks); - - InputPartition[] partitions = new InputPartition[combinedScanTasks.size()]; - - for (int index = 0; index < combinedScanTasks.size(); index++) { - partitions[index] = - new SparkInputPartition( - EMPTY_GROUPING_KEY_TYPE, - combinedScanTasks.get(index), - tableBroadcast, - fileIOBroadcast, - projection, - caseSensitive, - locations != null ? locations[index] : SparkPlanningUtil.NO_LOCATION_PREFERENCE, - cacheDeleteFilesOnExecutors); - } - - return partitions; - } - - private String[][] computePreferredLocations(List taskGroups) { - return localityPreferred - ? SparkPlanningUtil.fetchBlockLocations(fileIO.get(), taskGroups) - : null; - } - - @Override - public PartitionReaderFactory createReaderFactory() { - return new SparkRowReaderFactory(); - } - - @Override - public Offset initialOffset() { - return initialOffset; + return Lists.newArrayList( + TableScanUtil.planTasks(splitTasks, splitSize, splitLookback, splitOpenFileCost)); } @Override - public Offset deserializeOffset(String json) { - return StreamingOffset.fromJson(json); + protected boolean localityPreferred() { + return localityPreferred; } @Override - public void commit(Offset end) {} - - @Override - public void stop() { + protected void stopStream() { if (planner != null) { planner.stop(); } } private void initializePlanner(StreamingOffset startOffset, StreamingOffset endOffset) { - if (readConf.asyncMicroBatchPlanningEnabled()) { + if (readConf().asyncMicroBatchPlanningEnabled()) { this.planner = new AsyncSparkMicroBatchPlanner( - table, readConf, startOffset, endOffset, lastOffsetForTriggerAvailableNow); + table(), readConf(), startOffset, endOffset, lastOffsetForTriggerAvailableNow()); } else { this.planner = - new SyncSparkMicroBatchPlanner(table, readConf, lastOffsetForTriggerAvailableNow); + new SyncSparkMicroBatchPlanner(table(), readConf(), lastOffsetForTriggerAvailableNow()); } } @@ -249,69 +171,22 @@ public ReadLimit getDefaultReadLimit() { } @Override - public void prepareForTriggerAvailableNow() { + protected StreamingOffset availableNowEndOffset() { LOG.info("The streaming query reports to use Trigger.AvailableNow"); - lastOffsetForTriggerAvailableNow = - (StreamingOffset) latestOffset(initialOffset, ReadLimit.allAvailable()); + StreamingOffset endOffset = + (StreamingOffset) latestOffset(initialStreamingOffset(), ReadLimit.allAvailable()); - LOG.info("lastOffset for Trigger.AvailableNow is {}", lastOffsetForTriggerAvailableNow.json()); + LOG.info("lastOffset for Trigger.AvailableNow is {}", endOffset.json()); + return endOffset; + } + @Override + protected void availableNowPrepared() { // Reset planner so it gets recreated with the cap on next call if (planner != null) { planner.stop(); planner = null; } } - - private static class InitialOffsetStore { - private final Table table; - private final FileIO io; - private final String initialOffsetLocation; - private final long fromTimestamp; - - InitialOffsetStore( - Table table, String checkpointLocation, long fromTimestamp, Configuration conf) { - this.table = table; - this.io = new HadoopFileIO(conf); - this.initialOffsetLocation = SLASH.join(checkpointLocation, "offsets/0"); - this.fromTimestamp = fromTimestamp; - } - - public StreamingOffset initialOffset() { - InputFile inputFile = io.newInputFile(initialOffsetLocation); - if (inputFile.exists()) { - return readOffset(inputFile); - } - - table.refresh(); - StreamingOffset offset = MicroBatchUtils.determineStartingOffset(table, fromTimestamp); - - OutputFile outputFile = io.newOutputFile(initialOffsetLocation); - writeOffset(offset, outputFile); - - return offset; - } - - private void writeOffset(StreamingOffset offset, OutputFile file) { - try (OutputStream outputStream = file.create()) { - BufferedWriter writer = - new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8)); - writer.write(offset.json()); - writer.flush(); - } catch (IOException ioException) { - throw new UncheckedIOException( - String.format("Failed writing offset to: %s", initialOffsetLocation), ioException); - } - } - - private StreamingOffset readOffset(InputFile file) { - try (InputStream in = file.newStream()) { - return StreamingOffset.fromJson(in); - } catch (IOException ioException) { - throw new UncheckedIOException( - String.format("Failed reading offset from: %s", initialOffsetLocation), ioException); - } - } - } } diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStreamBase.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStreamBase.java new file mode 100644 index 000000000000..a063116d52ea --- /dev/null +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStreamBase.java @@ -0,0 +1,189 @@ +/* + * 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.iceberg.spark.source; + +import java.util.List; +import java.util.function.Supplier; +import org.apache.iceberg.ScanTaskGroup; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.spark.SparkReadConf; +import org.apache.iceberg.types.Types; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.broadcast.Broadcast; +import org.apache.spark.sql.connector.read.InputPartition; +import org.apache.spark.sql.connector.read.PartitionReaderFactory; +import org.apache.spark.sql.connector.read.streaming.MicroBatchStream; +import org.apache.spark.sql.connector.read.streaming.Offset; +import org.apache.spark.sql.connector.read.streaming.SupportsTriggerAvailableNow; + +abstract class SparkMicroBatchStreamBase implements MicroBatchStream, SupportsTriggerAvailableNow { + + private static final Types.StructType EMPTY_GROUPING_KEY_TYPE = Types.StructType.of(); + + private final JavaSparkContext sparkContext; + private final Table table; + private final Supplier fileIO; + private final SparkReadConf readConf; + private final String projection; + private Broadcast
tableBroadcast = null; + private final Broadcast fileIOBroadcast; + private final StreamingOffset initialOffset; + private StreamingOffset lastOffsetForTriggerAvailableNow = null; + + SparkMicroBatchStreamBase( + JavaSparkContext sparkContext, + Table table, + Supplier fileIO, + SparkReadConf readConf, + Schema projection, + String checkpointLocation, + Supplier initialOffsetSupplier) { + this.sparkContext = sparkContext; + this.table = table; + this.fileIO = fileIO; + this.readConf = readConf; + this.projection = SchemaParser.toJson(projection); + this.fileIOBroadcast = sparkContext.broadcast(SerializableFileIOWithSize.wrap(fileIO.get())); + this.initialOffset = + new StreamingInitialOffsetStore( + checkpointLocation, sparkContext.hadoopConfiguration(), initialOffsetSupplier) + .initialOffset(); + } + + protected final Table table() { + return table; + } + + protected final JavaSparkContext sparkContext() { + return sparkContext; + } + + protected final FileIO fileIO() { + return fileIO.get(); + } + + protected final SparkReadConf readConf() { + return readConf; + } + + protected final StreamingOffset initialStreamingOffset() { + return initialOffset; + } + + protected final StreamingOffset lastOffsetForTriggerAvailableNow() { + return lastOffsetForTriggerAvailableNow; + } + + @Override + public final Offset latestOffset() { + return lastOffsetForTriggerAvailableNow != null + ? lastOffsetForTriggerAvailableNow + : latestStreamingOffset(); + } + + protected abstract StreamingOffset latestStreamingOffset(); + + @Override + public final InputPartition[] planInputPartitions(Offset start, Offset end) { + Preconditions.checkArgument( + start instanceof StreamingOffset, "Invalid start offset: %s", start); + Preconditions.checkArgument(end instanceof StreamingOffset, "Invalid end offset: %s", end); + + List> taskGroups = + planTaskGroups((StreamingOffset) start, (StreamingOffset) end); + if (taskGroups.isEmpty()) { + return new InputPartition[0]; + } + + String[][] locations = + localityPreferred() ? SparkPlanningUtil.fetchBlockLocations(fileIO(), taskGroups) : null; + Broadcast
currentTableBroadcast = tableBroadcast(); + InputPartition[] partitions = new InputPartition[taskGroups.size()]; + for (int index = 0; index < taskGroups.size(); index++) { + partitions[index] = + new SparkInputPartition( + EMPTY_GROUPING_KEY_TYPE, + taskGroups.get(index), + currentTableBroadcast, + fileIOBroadcast, + projection, + readConf.caseSensitive(), + locations != null ? locations[index] : SparkPlanningUtil.NO_LOCATION_PREFERENCE, + readConf.cacheDeleteFilesOnExecutors()); + } + + return partitions; + } + + protected abstract List> planTaskGroups( + StreamingOffset startOffset, StreamingOffset endOffset); + + protected boolean localityPreferred() { + return false; + } + + protected Broadcast
tableBroadcast() { + if (tableBroadcast == null) { + this.tableBroadcast = sparkContext.broadcast(SerializableTableWithSize.copyOf(table)); + } + + return tableBroadcast; + } + + @Override + public final PartitionReaderFactory createReaderFactory() { + return new SparkRowReaderFactory(); + } + + @Override + public final Offset initialOffset() { + return initialOffset; + } + + @Override + public final Offset deserializeOffset(String json) { + return StreamingOffset.fromJson(json); + } + + @Override + public final void commit(Offset end) {} + + @Override + public final void stop() { + stopStream(); + } + + protected void stopStream() {} + + @Override + public final void prepareForTriggerAvailableNow() { + this.lastOffsetForTriggerAvailableNow = availableNowEndOffset(); + availableNowPrepared(); + } + + protected StreamingOffset availableNowEndOffset() { + return latestStreamingOffset(); + } + + protected void availableNowPrepared() {} +} diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/StreamingInitialOffsetStore.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/StreamingInitialOffsetStore.java new file mode 100644 index 000000000000..9d34033c5ea5 --- /dev/null +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/StreamingInitialOffsetStore.java @@ -0,0 +1,79 @@ +/* + * 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.iceberg.spark.source; + +import java.io.BufferedWriter; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.function.Supplier; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.hadoop.HadoopFileIO; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Joiner; + +class StreamingInitialOffsetStore { + private static final Joiner SLASH = Joiner.on("/"); + + private final FileIO io; + private final String initialOffsetLocation; + private final Supplier offsetSupplier; + + StreamingInitialOffsetStore( + String checkpointLocation, Configuration conf, Supplier offsetSupplier) { + this.io = new HadoopFileIO(conf); + this.initialOffsetLocation = SLASH.join(checkpointLocation, "offsets/0"); + this.offsetSupplier = offsetSupplier; + } + + StreamingOffset initialOffset() { + InputFile inputFile = io.newInputFile(initialOffsetLocation); + if (inputFile.exists()) { + return readOffset(inputFile); + } + + StreamingOffset offset = offsetSupplier.get(); + writeOffset(offset, io.newOutputFile(initialOffsetLocation)); + return offset; + } + + private void writeOffset(StreamingOffset offset, OutputFile file) { + try (OutputStream outputStream = file.create(); + BufferedWriter writer = + new BufferedWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8))) { + writer.write(offset.json()); + writer.flush(); + } catch (IOException e) { + throw new UncheckedIOException("Failed writing offset to: " + initialOffsetLocation, e); + } + } + + private StreamingOffset readOffset(InputFile file) { + try (InputStream in = file.newStream()) { + return StreamingOffset.fromJson(in); + } catch (IOException e) { + throw new UncheckedIOException("Failed reading offset from: " + initialOffsetLocation, e); + } + } +} diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkChangelog.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkChangelog.java new file mode 100644 index 000000000000..4bfc2c517439 --- /dev/null +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkChangelog.java @@ -0,0 +1,192 @@ +/* + * 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.iceberg.spark.source; + +import static org.apache.iceberg.TestHelpers.row; +import static org.assertj.core.api.Assertions.assertThat; + +import org.apache.iceberg.Table; +import org.apache.iceberg.spark.SparkReadConf; +import org.apache.iceberg.spark.TestBaseWithCatalog; +import org.apache.spark.api.java.JavaSparkContext; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.connector.read.streaming.ReadLimit; +import org.apache.spark.sql.streaming.StreamingQuery; +import org.apache.spark.sql.streaming.Trigger; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; + +class TestSparkChangelog extends TestBaseWithCatalog { + + @AfterEach + void removeTable() { + sql("DROP TABLE IF EXISTS %s", tableName); + } + + @TestTemplate + void readsChangesUsingSparkCdcSyntax() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + tableName); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + long firstVersion = table.currentSnapshot().sequenceNumber(); + + sql("INSERT INTO %s VALUES (3, 'c')", tableName); + table.refresh(); + long secondVersion = table.currentSnapshot().sequenceNumber(); + + assertThat( + sql( + "SELECT id, data, _change_type, _commit_version " + + "FROM %s CHANGES FROM VERSION %d TO VERSION %d ORDER BY id", + tableName, firstVersion, secondVersion)) + .containsExactly( + row(1L, "a", "insert", firstVersion), + row(2L, "b", "insert", firstVersion), + row(3L, "c", "insert", secondVersion)); + } + + @TestTemplate + void readsCopyOnWriteChangesUsingSparkCdcSyntax() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + tableName); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName); + + sql("DELETE FROM %s WHERE id = 1", tableName); + Table table = validationCatalog.loadTable(tableIdent); + long deleteVersion = table.currentSnapshot().sequenceNumber(); + assertThat(table.currentSnapshot().deleteManifests(table.io())).isEmpty(); + + assertThat( + sql( + "SELECT id, data, _change_type, _commit_version " + + "FROM %s CHANGES FROM VERSION %d TO VERSION %d " + + "ORDER BY _change_type, id", + tableName, deleteVersion, deleteVersion)) + .containsExactly(row(1L, "a", "delete", deleteVersion)); + + assertThat( + sql( + "SELECT id, data, _change_type, _commit_version " + + "FROM %s CHANGES FROM VERSION %d TO VERSION %d " + + "WITH (deduplicationMode = 'none') " + + "ORDER BY _change_type, id", + tableName, deleteVersion, deleteVersion)) + .containsExactly( + row(1L, "a", "delete", deleteVersion), + row(2L, "b", "delete", deleteVersion), + row(2L, "b", "insert", deleteVersion)); + } + + @TestTemplate + void computesUpdatesForCopyOnWriteChanges() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + tableName); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName); + + sql("UPDATE %s SET data = 'updated' WHERE id = 1", tableName); + Table table = validationCatalog.loadTable(tableIdent); + long updateVersion = table.currentSnapshot().sequenceNumber(); + + assertThat( + sql( + "SELECT id, data, _change_type, _commit_version " + + "FROM %s CHANGES FROM VERSION %d TO VERSION %d " + + "WITH (computeUpdates = 'true') ORDER BY data", + tableName, updateVersion, updateVersion)) + .containsExactly( + row(1L, "a", "update_preimage", updateVersion), + row(1L, "updated", "update_postimage", updateVersion)); + } + + @TestTemplate + void availableNowPinsLatestSnapshot() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + tableName); + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + long availableSnapshotId = table.currentSnapshot().snapshotId(); + SparkReadConf readConf = new SparkReadConf(spark, table, CaseInsensitiveStringMap.empty()); + SparkChangelogMicroBatchStream stream = + new SparkChangelogMicroBatchStream( + JavaSparkContext.fromSparkContext(spark.sparkContext()), + table, + readConf, + SparkChangelogTable.cdcDataSchema(table), + temp.resolve("cdc-available-now").toString()); + + stream.prepareForTriggerAvailableNow(); + sql("INSERT INTO %s VALUES (2, 'b')", tableName); + + StreamingOffset latestOffset = (StreamingOffset) stream.latestOffset(); + assertThat(latestOffset.snapshotId()).isEqualTo(availableSnapshotId); + assertThat(latestOffset.position()).isZero(); + assertThat(stream.latestOffset(stream.initialOffset(), ReadLimit.allAvailable())) + .isEqualTo(latestOffset); + stream.stop(); + } + + @TestTemplate + void streamsChangesUsingSparkCdcApi() throws Exception { + String queryName = "iceberg_cdc_changes"; + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + tableName); + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName); + + Dataset changes = spark.readStream().changes(tableName); + StreamingQuery query = + changes + .writeStream() + .format("memory") + .queryName(queryName) + .trigger(Trigger.AvailableNow()) + .start(); + query.awaitTermination(); + + assertThat(sql("SELECT id, data, _change_type FROM %s ORDER BY id", queryName)) + .containsExactly(row(1L, "a", "insert"), row(2L, "b", "insert")); + spark.catalog().dropTempView(queryName); + } + + @TestTemplate + void tableChangesKeepsIcebergChangelogColumns() { + sql("CREATE TABLE %s (id bigint, data string) USING iceberg", tableName); + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + + assertThat(sql("SELECT id, data, _change_type, _commit_snapshot_id FROM %s.changes", tableName)) + .hasSize(1) + .allSatisfy( + row -> { + assertThat(row[2]).isEqualTo("INSERT"); + assertThat(row[3]).isInstanceOf(Long.class); + }); + } +} diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingInitialOffsetStore.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingInitialOffsetStore.java new file mode 100644 index 000000000000..dd8f2dca0ca3 --- /dev/null +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingInitialOffsetStore.java @@ -0,0 +1,60 @@ +/* + * 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.iceberg.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class TestStreamingInitialOffsetStore { + + @TempDir private Path checkpointDir; + + @Test + void restoresStoredOffset() { + AtomicInteger initializations = new AtomicInteger(); + StreamingOffset expected = new StreamingOffset(34L, 0L, false); + StreamingInitialOffsetStore firstStore = + new StreamingInitialOffsetStore( + checkpointDir.toString(), + new Configuration(), + () -> { + initializations.incrementAndGet(); + return expected; + }); + + assertThat(firstStore.initialOffset()).isEqualTo(expected); + + StreamingInitialOffsetStore restoredStore = + new StreamingInitialOffsetStore( + checkpointDir.toString(), + new Configuration(), + () -> { + initializations.incrementAndGet(); + return StreamingOffset.START_OFFSET; + }); + + assertThat(restoredStore.initialOffset()).isEqualTo(expected); + assertThat(initializations).hasValue(1); + } +}