You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sz...@apache.org on 2026/09/17 22:17:10 UTC
(iceberg) branch main updated: Spark 4.2: Extract streaming initial offset store (#18092)
This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 778d103c0f Spark 4.2: Extract streaming initial offset store (#18092)
778d103c0f is described below
commit 778d103c0fa802be5e85e4cb2e4e4b4bcd8ac891
Author: Xin Huang <42...@users.noreply.github.com>
AuthorDate: Thu Sep 17 15:17:04 2026 -0700
Spark 4.2: Extract streaming initial offset store (#18092)
---
.../spark/source/SparkMicroBatchStream.java | 75 ++---------
.../spark/source/StreamingInitialOffsetStore.java | 79 +++++++++++
.../source/TestStreamingInitialOffsetStore.java | 145 +++++++++++++++++++++
3 files changed, 232 insertions(+), 67 deletions(-)
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 7adf3c633c..2d510e50ad 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,28 +18,16 @@
*/
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;
@@ -57,7 +45,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerAvailableNow {
- private static final Joiner SLASH = Joiner.on("/");
private static final Logger LOG = LoggerFactory.getLogger(SparkMicroBatchStream.class);
private static final Types.StructType EMPTY_GROUPING_KEY_TYPE = Types.StructType.of();
@@ -103,9 +90,14 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerA
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
this.cacheDeleteFilesOnExecutors = readConf.cacheDeleteFilesOnExecutors();
- InitialOffsetStore initialOffsetStore =
- new InitialOffsetStore(
- table, checkpointLocation, fromTimestamp, sparkContext.hadoopConfiguration());
+ StreamingInitialOffsetStore initialOffsetStore =
+ new StreamingInitialOffsetStore(
+ checkpointLocation,
+ sparkContext.hadoopConfiguration(),
+ () -> {
+ table.refresh();
+ return MicroBatchUtils.determineStartingOffset(table, fromTimestamp);
+ });
this.initialOffset = initialOffsetStore.initialOffset();
}
@@ -263,55 +255,4 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerA
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/StreamingInitialOffsetStore.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/StreamingInitialOffsetStore.java
new file mode 100644
index 0000000000..9d34033c5e
--- /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<StreamingOffset> offsetSupplier;
+
+ StreamingInitialOffsetStore(
+ String checkpointLocation, Configuration conf, Supplier<StreamingOffset> 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/TestStreamingInitialOffsetStore.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingInitialOffsetStore.java
new file mode 100644
index 0000000000..d14e4c3b77
--- /dev/null
+++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingInitialOffsetStore.java
@@ -0,0 +1,145 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestStreamingInitialOffsetStore {
+
+ @TempDir private Path checkpointDir;
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void persistsOffsetAndCreatesParentDirectories(boolean scanAllFiles) {
+ Path checkpoint = checkpointDir.resolve("nested/checkpoint");
+ StreamingOffset expected = new StreamingOffset(34L, 7L, scanAllFiles);
+ StreamingInitialOffsetStore store =
+ new StreamingInitialOffsetStore(checkpoint.toString(), new Configuration(), () -> expected);
+
+ assertThat(store.initialOffset()).isEqualTo(expected);
+ assertThat(checkpoint.resolve("offsets/0"))
+ .isRegularFile()
+ .hasContent(
+ "{\"version\":1,\"snapshot_id\":34,\"position\":7,\"scan_all_files\":%s}"
+ .formatted(scanAllFiles));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void restoresStoredOffsetWithoutInitializing(boolean scanAllFiles) throws IOException {
+ Path offsetFile = checkpointDir.resolve("offsets/0");
+ Files.createDirectories(offsetFile.getParent());
+ Files.writeString(
+ offsetFile,
+ "{\"version\":1,\"snapshot_id\":34,\"position\":7,\"scan_all_files\":%s}"
+ .formatted(scanAllFiles));
+ StreamingInitialOffsetStore store =
+ new StreamingInitialOffsetStore(
+ checkpointDir.toString(),
+ new Configuration(),
+ () -> {
+ throw new AssertionError("Must restore the persisted offset without initializing");
+ });
+
+ assertThat(store.initialOffset()).isEqualTo(new StreamingOffset(34L, 7L, scanAllFiles));
+ }
+
+ @ParameterizedTest
+ @MethodSource("invalidOffsets")
+ void rejectsInvalidCheckpointWithoutReinitializing(
+ String json, Class<? extends Throwable> exceptionClass, String message) throws IOException {
+ Path offsetFile = checkpointDir.resolve("offsets/0");
+ Files.createDirectories(offsetFile.getParent());
+ Files.writeString(offsetFile, json);
+ StreamingInitialOffsetStore store =
+ new StreamingInitialOffsetStore(
+ checkpointDir.toString(),
+ new Configuration(),
+ () -> {
+ throw new AssertionError("Must not replace an invalid checkpoint with a new offset");
+ });
+
+ assertThatThrownBy(store::initialOffset)
+ .isInstanceOf(exceptionClass)
+ .hasMessageContaining(message);
+ assertThat(offsetFile).hasContent(json);
+ }
+
+ private static Stream<Arguments> invalidOffsets() {
+ return Stream.of(
+ Arguments.of(
+ "{\"version\":2,\"snapshot_id\":34,\"position\":7,\"scan_all_files\":false}",
+ IllegalArgumentException.class,
+ "Version 2 is not supported"),
+ Arguments.of(
+ "{\"version\":\"foo\",\"snapshot_id\":34,\"position\":7,\"scan_all_files\":false}",
+ IllegalArgumentException.class,
+ "Cannot parse to an integer value: version"),
+ Arguments.of(
+ "{\"snapshot_id\":34,\"position\":7,\"scan_all_files\":false}",
+ IllegalArgumentException.class,
+ "Cannot parse missing int: version"),
+ Arguments.of(
+ "{\"version\":1",
+ UncheckedIOException.class,
+ "Failed to read StreamingOffset from json"));
+ }
+
+ @Test
+ void restoresStartOffsetWithoutReinitializing() {
+ AtomicInteger initializations = new AtomicInteger();
+ StreamingInitialOffsetStore firstStore =
+ new StreamingInitialOffsetStore(
+ checkpointDir.toString(),
+ new Configuration(),
+ () -> {
+ initializations.incrementAndGet();
+ return StreamingOffset.START_OFFSET;
+ });
+
+ assertThat(firstStore.initialOffset()).isEqualTo(StreamingOffset.START_OFFSET);
+
+ StreamingInitialOffsetStore restoredStore =
+ new StreamingInitialOffsetStore(
+ checkpointDir.toString(),
+ new Configuration(),
+ () -> {
+ initializations.incrementAndGet();
+ return new StreamingOffset(34L, 0L, false);
+ });
+
+ assertThat(restoredStore.initialOffset()).isEqualTo(StreamingOffset.START_OFFSET);
+ assertThat(initializations).hasValue(1);
+ }
+}