You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "c2zwdjnlcg (via GitHub)" <gi...@apache.org> on 2026/02/05 17:33:09 UTC
[PR] spark: Don't use table FileIO for checkpointing files [iceberg]
c2zwdjnlcg opened a new pull request, #15239:
URL: https://github.com/apache/iceberg/pull/15239
Fixes: #14762
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2897305213
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -95,8 +96,9 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerA
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
this.cacheDeleteFilesOnExecutors = readConf.cacheDeleteFilesOnExecutors();
Review Comment:
Kind of a nit but we can be a little cleaner in intent now:
Initial Offset only really needs HadoopConfiguration now so we may as well make the HadoopFileIO there, also lets us change this later if we really want to use HadoopFS
So basically InitialOffsetStore(table, checkpointLocation, fromTimestamp, Configuration)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "c2zwdjnlcg (via GitHub)" <gi...@apache.org>.
c2zwdjnlcg commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2893011105
##########
docs/docs/spark-configuration.md:
##########
@@ -220,6 +220,7 @@ spark.read
| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per microbatch |
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit |
+| streaming-checkpoint-use-hadoop | false | Use Hadoop FileSystem for streaming checkpoint operations instead of the table's FileIO implementation |
Review Comment:
@RussellSpitzer made the change, let me know if you want me to squash
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2854987230
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStreamingCheckpointLocation.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.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+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.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to verify that streaming checkpoint configuration correctly controls which I/O
+ * implementation is used for checkpoint operations.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestStreamingCheckpointLocation extends TestBaseWithCatalog {
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ "testtableio",
+ SparkCatalog.class.getName(),
+ ImmutableMap.of(
+ "type",
+ "hadoop",
+ CatalogProperties.FILE_IO_IMPL,
+ TrackingFileIO.class.getName(),
+ "cache-enabled",
+ "false")
+ }
+ };
+ }
+
+ @AfterEach
+ public void stopStreams() throws TimeoutException {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ TrackingFileIO.reset();
+ }
+
+ @TestTemplate
+ public void testTableIOMode() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Use nested checkpoint path to verify parent directory handling
+ File checkpointDir = new File(temp.toFile(), "nested/checkpoint");
+ TrackingFileIO.reset();
+
+ // Default: streaming-checkpoint-use-table-io defaults to true
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ query.processAllAvailable();
+ query.stop();
+
+ // Verify TrackingFileIO was used
+ assertThat(TrackingFileIO.wasUsed())
+ .as("Table FileIO should be used for checkpoints when config is true/default")
+ .isTrue();
+
+ assertThat(new File(checkpointDir, "offsets/0")).exists().isFile();
+ }
+
+ @TestTemplate
+ public void testHadoopFSMode() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Use nested checkpoint path to verify parent directory creation
+ File checkpointDir = new File(temp.toFile(), "deeply/nested/checkpoint");
+ TrackingFileIO.reset();
+
+ // Explicit: streaming-checkpoint-use-table-io=false (uses Hadoop FS)
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .option("streaming-checkpoint-use-table-io", "false")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ query.processAllAvailable();
+ query.stop();
+
+ // Verify TrackingFileIO was NOT used
+ assertThat(TrackingFileIO.wasUsed())
+ .as("Hadoop FS should bypass table FileIO when config is false")
+ .isFalse();
+
+ // Verify Hadoop FS created the nested directory structure
+ assertThat(new File(checkpointDir, "offsets/0")).exists().isFile();
+ }
+
+ /**
+ * A FileIO that tracks whether it was used for checkpoint operations. This allows us to verify
+ * which I/O path (TableIO vs Hadoop FS) is being used.
+ */
+ public static class TrackingFileIO implements FileIO {
+ private static final AtomicBoolean USED = new AtomicBoolean(false);
+ private FileIO delegate;
+
+ public static void reset() {
+ USED.set(false);
+ }
+
+ public static boolean wasUsed() {
+ return USED.get();
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ if (path.contains("/offsets/")) {
+ USED.set(true);
+ }
+ return delegate.newInputFile(path);
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ if (path.contains("/offsets/")) {
Review Comment:
nit: could use a constant here and in the tests above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-3953406887
First look, please do larger changes like this only on a single module first, then backport to the others in a follow up. It makes reviewing a bit more difficult to have duplicated changes. Taking a pass in depth now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "c2zwdjnlcg (via GitHub)" <gi...@apache.org>.
c2zwdjnlcg commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2897399830
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -95,8 +96,9 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsTriggerA
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
this.cacheDeleteFilesOnExecutors = readConf.cacheDeleteFilesOnExecutors();
Review Comment:
updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2854689938
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -247,52 +267,147 @@ public void prepareForTriggerAvailableNow() {
}
}
- private static class InitialOffsetStore {
+ private abstract static class BaseOffsetStore {
+ private static final Joiner SLASH = Joiner.on("/");
+
private final Table table;
- private final FileIO io;
private final String initialOffsetLocation;
private final Long fromTimestamp;
- InitialOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
+ BaseOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
this.table = table;
- this.io = table.io();
this.initialOffsetLocation = SLASH.join(checkpointLocation, "offsets/0");
this.fromTimestamp = fromTimestamp;
}
+ protected Table table() {
+ return table;
+ }
+
+ protected String initialOffsetLocation() {
+ return initialOffsetLocation;
+ }
+
+ protected Long fromTimestamp() {
+ return fromTimestamp;
+ }
+
public StreamingOffset initialOffset() {
- InputFile inputFile = io.newInputFile(initialOffsetLocation);
- if (inputFile.exists()) {
- return readOffset(inputFile);
+ if (offsetExists()) {
+ return readOffset();
}
- table.refresh();
- StreamingOffset offset = MicroBatchUtils.determineStartingOffset(table, fromTimestamp);
+ table().refresh();
+ StreamingOffset offset = MicroBatchUtils.determineStartingOffset(table(), fromTimestamp());
+ writeOffset(offset);
+ return offset;
+ }
- OutputFile outputFile = io.newOutputFile(initialOffsetLocation);
- writeOffset(offset, outputFile);
+ protected abstract boolean offsetExists();
- return offset;
+ protected abstract StreamingOffset readOffset();
+
+ protected abstract void writeOffset(StreamingOffset offset);
+ }
+
+ private static class TableIOOffsetStore extends BaseOffsetStore {
+ private final FileIO io;
+
+ TableIOOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
+ super(table, checkpointLocation, fromTimestamp);
+ this.io = table.io();
}
- private void writeOffset(StreamingOffset offset, OutputFile file) {
+ @Override
+ protected boolean offsetExists() {
+ InputFile inputFile = io.newInputFile(initialOffsetLocation());
+ return inputFile.exists();
+ }
+
+ @Override
+ protected StreamingOffset readOffset() {
+ InputFile file = io.newInputFile(initialOffsetLocation());
+ try (InputStream in = file.newStream()) {
+ return StreamingOffset.fromJson(in);
+ } catch (IOException ioException) {
+ throw new UncheckedIOException(
+ String.format("Failed reading offset from: %s", initialOffsetLocation()), ioException);
+ }
+ }
+
+ @Override
+ protected void writeOffset(StreamingOffset offset) {
+ OutputFile file = io.newOutputFile(initialOffsetLocation());
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);
+ String.format("Failed writing offset to: %s", initialOffsetLocation()), ioException);
}
}
+ }
- private StreamingOffset readOffset(InputFile file) {
- try (InputStream in = file.newStream()) {
- return StreamingOffset.fromJson(in);
+ private static class HadoopOffsetStore extends BaseOffsetStore {
+ private final FileSystem fileSystem;
+ private final Path initialOffsetPath;
+
+ HadoopOffsetStore(
+ Table table, String checkpointLocation, Long fromTimestamp, Configuration hadoopConf) {
+ super(table, checkpointLocation, fromTimestamp);
+ this.initialOffsetPath = new Path(initialOffsetLocation());
+
+ try {
+ this.fileSystem = initialOffsetPath.getFileSystem(hadoopConf);
+ } catch (IOException ioException) {
+ throw new UncheckedIOException(
+ String.format("Failed to get FileSystem for: %s", initialOffsetPath), ioException);
+ }
+ }
+
+ @Override
+ protected boolean offsetExists() {
+ try {
+ return fileSystem.exists(initialOffsetPath);
+ } catch (IOException ioException) {
+ throw new UncheckedIOException(
+ String.format("Failed to check existence of: %s", initialOffsetPath), ioException);
+ }
+ }
+
+ @Override
+ protected StreamingOffset readOffset() {
+ try (FSDataInputStream inputStream = fileSystem.open(initialOffsetPath);
+ InputStreamReader reader = new InputStreamReader(inputStream, StandardCharsets.UTF_8)) {
+ String json = CharStreams.toString(reader);
Review Comment:
Don't we already have a StreamingOffset.fromJson(InputStream in) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-4027887985
Merged Thanks @c2zwdjnlcg , and thanks to @danielcweeks and @nastra for their reviews. @c2zwdjnlcg ping me when the backports are ready and we can clean up the previous spark modules
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2848359084
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -253,9 +255,15 @@ private static class InitialOffsetStore {
private final String initialOffsetLocation;
private final Long fromTimestamp;
- InitialOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
+ InitialOffsetStore(
+ Table table, String checkpointLocation, Long fromTimestamp, JavaSparkContext sparkContext) {
this.table = table;
- this.io = table.io();
+ // Use ResolvingFileIO with empty properties to get defaults for the checkpoint location
+ this.io =
+ CatalogUtil.loadFileIO(
+ "org.apache.iceberg.io.ResolvingFileIO",
Review Comment:
Using `Collections.emptyMap()` discards all user configuration. If the checkpoint location is on S3 with a custom endpoint, custom credential provider, or regional configuration, this won't pick any of that up. The empty properties are the main concern here — this is essentially creating an unconfigured FileIO and hoping the Hadoop config alone is enough.
Also, hard-coding `"org.apache.iceberg.io.ResolvingFileIO"` creates a tight coupling to a specific implementation. `ResolvingFileIO` is currently not used anywhere in the Spark module, and its fallback is `HadoopFileIO`, which means this is effectively a Hadoop dependency in disguise.
I'd suggest either:
1. Using Hadoop `FileSystem` directly (since we're already calling `sparkContext.hadoopConfiguration()` here anyway), or
2. If staying within FileIO, using `HadoopFileIO` directly — which is simpler, more explicit, and what `ResolvingFileIO` would resolve to for most checkpoint path schemes (local, hdfs).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer merged PR #15239:
URL: https://github.com/apache/iceberg/pull/15239
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2854863050
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java:
##########
@@ -84,6 +84,11 @@ private SparkReadOptions() {}
public static final String STREAMING_MAX_ROWS_PER_MICRO_BATCH =
"streaming-max-rows-per-micro-batch";
+ // Controls whether streaming checkpoint operations use table FileIO or Hadoop FileSystem
+ public static final String STREAMING_CHECKPOINT_USE_TABLE_IO =
+ "streaming-checkpoint-use-table-io";
Review Comment:
I'm not sure if it should be use-table-io or use-hdfs ... I think either is probably fine but maybe I slightly prefer use-hdfs because I know the opposite should be using the table io?
I do think using an enum may be overboard here but maybe that's just cleaner all around
streaming-checkpoint = {table-io, hdfs}
streaming-checkpoint_default = table-io
?
Wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "c2zwdjnlcg (via GitHub)" <gi...@apache.org>.
c2zwdjnlcg commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2874284156
##########
docs/docs/spark-configuration.md:
##########
@@ -220,6 +220,7 @@ spark.read
| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per microbatch |
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit |
+| streaming-checkpoint-use-hadoop | false | Use Hadoop FileSystem for streaming checkpoint operations instead of the table's FileIO implementation |
Review Comment:
Makes sense to me, but @danielcweeks on you onboard with that? If so I can make the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2848359088
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCheckpointFileIO.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.io.File;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Table;
+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.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to verify that checkpoint locations use the correct FileIO configuration, independent of
+ * the table's FileIO configuration.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCheckpointFileIO extends TestBaseWithCatalog {
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ "testhive",
+ SparkCatalog.class.getName(),
+ ImmutableMap.of(
+ "type",
+ "hive",
+ CatalogProperties.FILE_IO_IMPL,
+ RestrictedPathFileIO.class.getName(),
+ "default-namespace",
+ "default")
+ },
+ };
+ }
+
+ @AfterEach
+ public void stopStreams() throws TimeoutException {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ RestrictedPathFileIO.setRejectedPrefix(null);
+ }
+
+ @TestTemplate
+ public void testCheckpointLocationDoesNotUseTableFileIO() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Verify the table is using RestrictedPathFileIO from the Spark catalog
+ SparkCatalog sparkCatalog =
+ (SparkCatalog) spark.sessionState().catalogManager().catalog(catalogName);
+ Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
+ SparkTable sparkTable = (SparkTable) sparkCatalog.loadTable(identifier);
+ Table table = sparkTable.table();
+ assertThat(table.io())
+ .as("Table should be using RestrictedPathFileIO from catalog configuration")
+ .isInstanceOf(RestrictedPathFileIO.class);
+
+ File checkpointDir = new File(temp.toFile(), "checkpoint");
+
+ // Configure RestrictedPathFileIO to reject this specific checkpoint path
+ // Include file:// scheme to match the actual paths being validated
+ RestrictedPathFileIO.setRejectedPrefix("file:" + checkpointDir.getAbsolutePath());
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ try {
+ query.processAllAvailable();
+
+ File offsetsDir = new File(checkpointDir, "offsets");
+ assertThat(offsetsDir).exists().isDirectory();
+ assertThat(new File(offsetsDir, "0")).exists().isFile();
+ } finally {
+ query.stop();
+ }
+ }
+
+ /**
+ * A FileIO implementation that rejects operations on paths with a specific prefix. This is used
+ * to test that checkpoint locations don't use the table's FileIO configuration.
+ */
+ public static class RestrictedPathFileIO implements FileIO {
+ private static String rejectedPrefix;
+ private FileIO delegate;
+
+ public static void setRejectedPrefix(String prefix) {
+ rejectedPrefix = prefix;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ validatePath(path);
+ return delegate.newInputFile(path);
+ }
+
+ @Override
Review Comment:
The `RestrictedPathFileIO` uses `static` mutable state (`rejectedPrefix`) which can cause test isolation issues in parallel test execution. Consider using an instance-based approach or test-specific configuration via properties.
Also, this test only exercises a local file path checkpoint. It would be good to have coverage for the case where the checkpoint location uses a cloud scheme (s3://, gs://, etc.) to verify the new IO actually works for those scenarios — that's where the original bug manifests.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCheckpointFileIO.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.io.File;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Table;
+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.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to verify that checkpoint locations use the correct FileIO configuration, independent of
+ * the table's FileIO configuration.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCheckpointFileIO extends TestBaseWithCatalog {
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ "testhive",
+ SparkCatalog.class.getName(),
+ ImmutableMap.of(
+ "type",
+ "hive",
+ CatalogProperties.FILE_IO_IMPL,
+ RestrictedPathFileIO.class.getName(),
+ "default-namespace",
+ "default")
+ },
+ };
+ }
+
+ @AfterEach
+ public void stopStreams() throws TimeoutException {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ RestrictedPathFileIO.setRejectedPrefix(null);
+ }
+
+ @TestTemplate
+ public void testCheckpointLocationDoesNotUseTableFileIO() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Verify the table is using RestrictedPathFileIO from the Spark catalog
+ SparkCatalog sparkCatalog =
+ (SparkCatalog) spark.sessionState().catalogManager().catalog(catalogName);
+ Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
+ SparkTable sparkTable = (SparkTable) sparkCatalog.loadTable(identifier);
+ Table table = sparkTable.table();
+ assertThat(table.io())
+ .as("Table should be using RestrictedPathFileIO from catalog configuration")
+ .isInstanceOf(RestrictedPathFileIO.class);
+
+ File checkpointDir = new File(temp.toFile(), "checkpoint");
+
+ // Configure RestrictedPathFileIO to reject this specific checkpoint path
+ // Include file:// scheme to match the actual paths being validated
+ RestrictedPathFileIO.setRejectedPrefix("file:" + checkpointDir.getAbsolutePath());
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ try {
+ query.processAllAvailable();
+
+ File offsetsDir = new File(checkpointDir, "offsets");
+ assertThat(offsetsDir).exists().isDirectory();
+ assertThat(new File(offsetsDir, "0")).exists().isFile();
+ } finally {
+ query.stop();
+ }
+ }
+
+ /**
+ * A FileIO implementation that rejects operations on paths with a specific prefix. This is used
+ * to test that checkpoint locations don't use the table's FileIO configuration.
+ */
+ public static class RestrictedPathFileIO implements FileIO {
+ private static String rejectedPrefix;
+ private FileIO delegate;
+
+ public static void setRejectedPrefix(String prefix) {
+ rejectedPrefix = prefix;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ validatePath(path);
+ return delegate.newInputFile(path);
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ validatePath(path);
+ return delegate.newOutputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ validatePath(path);
+ delegate.deleteFile(path);
+ }
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ this.delegate = new HadoopFileIO();
+ this.delegate.initialize(properties);
Review Comment:
The `RestrictedPathFileIO.initialize()` delegates to `new HadoopFileIO()` — this means the test itself depends on `HadoopFileIO`. If the goal is to verify checkpoint IO independence from the table FileIO, the test should ideally not hardcode the delegate to be Hadoop-based either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-3970315187
```
* What went wrong:
Execution failed for task ':iceberg-spark:iceberg-spark-4.1_2.13:spotlessJavaCheck'.
> The following files had format violations:
src/test/java/org/apache/iceberg/spark/source/TestStreamingCheckpointLocation.java
@@ -27,14 +27,13 @@
import·org.apache.iceberg.CatalogProperties;
import·org.apache.iceberg.ParameterizedTestExtension;
import·org.apache.iceberg.Parameters;
+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.hadoop.HadoopFileIO;
import·org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import·org.apache.iceberg.spark.SparkCatalog;
import·org.apache.iceberg.spark.SparkReadConf;
-import·org.apache.iceberg.spark.SparkReadOptions;
import·org.apache.iceberg.spark.TestBaseWithCatalog;
import·org.apache.spark.sql.streaming.StreamingQuery;
import·org.junit.jupiter.api.AfterEach;
```
**Run './gradlew spotlessApply' to fix all violations.**
We are going to be good to merge after all the tests pass. If you can just add addtional commits in the future rather than squashing and force pushing. It's easier to check the delta that way, the reviewer will always squash on merge so it's not a big deal
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "c2zwdjnlcg (via GitHub)" <gi...@apache.org>.
c2zwdjnlcg commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-3931014835
@nastra could you take a look at this PR and see if you are aligned with separating the checkpoint IO from the table IO?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2848359084
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -253,9 +255,15 @@ private static class InitialOffsetStore {
private final String initialOffsetLocation;
private final Long fromTimestamp;
- InitialOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
+ InitialOffsetStore(
+ Table table, String checkpointLocation, Long fromTimestamp, JavaSparkContext sparkContext) {
this.table = table;
- this.io = table.io();
+ // Use ResolvingFileIO with empty properties to get defaults for the checkpoint location
+ this.io =
+ CatalogUtil.loadFileIO(
+ "org.apache.iceberg.io.ResolvingFileIO",
Review Comment:
Using `Collections.emptyMap()` discards all user configuration. If the checkpoint location is on S3 with a custom endpoint, custom credential provider, or regional configuration, this won't pick any of that up. The empty properties are the main concern here — this is essentially creating an unconfigured FileIO and hoping the Hadoop config alone is enough.
Also, hard-coding `"org.apache.iceberg.io.ResolvingFileIO"` creates a tight coupling to a specific implementation. `ResolvingFileIO` is currently not used anywhere in the Spark module, and its fallback is `HadoopFileIO`, which means this is effectively a Hadoop dependency in disguise.
I'd suggest either:
1. Using Hadoop `FileSystem` directly (since we're already calling `sparkContext.hadoopConfiguration()` here anyway), or
2. If staying within FileIO, using `HadoopFileIO` directly — which is simpler, more explicit, and what `ResolvingFileIO` would resolve to for most checkpoint path schemes (local, hdfs).
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCheckpointFileIO.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.io.File;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Table;
+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.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to verify that checkpoint locations use the correct FileIO configuration, independent of
+ * the table's FileIO configuration.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCheckpointFileIO extends TestBaseWithCatalog {
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ "testhive",
+ SparkCatalog.class.getName(),
+ ImmutableMap.of(
+ "type",
+ "hive",
+ CatalogProperties.FILE_IO_IMPL,
+ RestrictedPathFileIO.class.getName(),
+ "default-namespace",
+ "default")
+ },
+ };
+ }
+
+ @AfterEach
+ public void stopStreams() throws TimeoutException {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ RestrictedPathFileIO.setRejectedPrefix(null);
+ }
+
+ @TestTemplate
+ public void testCheckpointLocationDoesNotUseTableFileIO() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Verify the table is using RestrictedPathFileIO from the Spark catalog
+ SparkCatalog sparkCatalog =
+ (SparkCatalog) spark.sessionState().catalogManager().catalog(catalogName);
+ Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
+ SparkTable sparkTable = (SparkTable) sparkCatalog.loadTable(identifier);
+ Table table = sparkTable.table();
+ assertThat(table.io())
+ .as("Table should be using RestrictedPathFileIO from catalog configuration")
+ .isInstanceOf(RestrictedPathFileIO.class);
+
+ File checkpointDir = new File(temp.toFile(), "checkpoint");
+
+ // Configure RestrictedPathFileIO to reject this specific checkpoint path
+ // Include file:// scheme to match the actual paths being validated
+ RestrictedPathFileIO.setRejectedPrefix("file:" + checkpointDir.getAbsolutePath());
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ try {
+ query.processAllAvailable();
+
+ File offsetsDir = new File(checkpointDir, "offsets");
+ assertThat(offsetsDir).exists().isDirectory();
+ assertThat(new File(offsetsDir, "0")).exists().isFile();
+ } finally {
+ query.stop();
+ }
+ }
+
+ /**
+ * A FileIO implementation that rejects operations on paths with a specific prefix. This is used
+ * to test that checkpoint locations don't use the table's FileIO configuration.
+ */
+ public static class RestrictedPathFileIO implements FileIO {
+ private static String rejectedPrefix;
+ private FileIO delegate;
+
+ public static void setRejectedPrefix(String prefix) {
+ rejectedPrefix = prefix;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ validatePath(path);
+ return delegate.newInputFile(path);
+ }
+
+ @Override
Review Comment:
The `RestrictedPathFileIO` uses `static` mutable state (`rejectedPrefix`) which can cause test isolation issues in parallel test execution. Consider using an instance-based approach or test-specific configuration via properties.
Also, this test only exercises a local file path checkpoint. It would be good to have coverage for the case where the checkpoint location uses a cloud scheme (s3://, gs://, etc.) to verify the new IO actually works for those scenarios — that's where the original bug manifests.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCheckpointFileIO.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.io.File;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Table;
+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.collect.ImmutableMap;
+import org.apache.iceberg.spark.SparkCatalog;
+import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests to verify that checkpoint locations use the correct FileIO configuration, independent of
+ * the table's FileIO configuration.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCheckpointFileIO extends TestBaseWithCatalog {
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}")
+ protected static Object[][] parameters() {
+ return new Object[][] {
+ {
+ "testhive",
+ SparkCatalog.class.getName(),
+ ImmutableMap.of(
+ "type",
+ "hive",
+ CatalogProperties.FILE_IO_IMPL,
+ RestrictedPathFileIO.class.getName(),
+ "default-namespace",
+ "default")
+ },
+ };
+ }
+
+ @AfterEach
+ public void stopStreams() throws TimeoutException {
+ for (StreamingQuery query : spark.streams().active()) {
+ query.stop();
+ }
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ RestrictedPathFileIO.setRejectedPrefix(null);
+ }
+
+ @TestTemplate
+ public void testCheckpointLocationDoesNotUseTableFileIO() throws Exception {
+ sql("CREATE TABLE %s (id INT, data STRING) USING iceberg", tableName);
+ sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b')", tableName);
+
+ // Verify the table is using RestrictedPathFileIO from the Spark catalog
+ SparkCatalog sparkCatalog =
+ (SparkCatalog) spark.sessionState().catalogManager().catalog(catalogName);
+ Identifier identifier = Identifier.of(tableIdent.namespace().levels(), tableIdent.name());
+ SparkTable sparkTable = (SparkTable) sparkCatalog.loadTable(identifier);
+ Table table = sparkTable.table();
+ assertThat(table.io())
+ .as("Table should be using RestrictedPathFileIO from catalog configuration")
+ .isInstanceOf(RestrictedPathFileIO.class);
+
+ File checkpointDir = new File(temp.toFile(), "checkpoint");
+
+ // Configure RestrictedPathFileIO to reject this specific checkpoint path
+ // Include file:// scheme to match the actual paths being validated
+ RestrictedPathFileIO.setRejectedPrefix("file:" + checkpointDir.getAbsolutePath());
+ StreamingQuery query =
+ spark
+ .readStream()
+ .format("iceberg")
+ .load(tableName)
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", checkpointDir.getAbsolutePath())
+ .start();
+
+ try {
+ query.processAllAvailable();
+
+ File offsetsDir = new File(checkpointDir, "offsets");
+ assertThat(offsetsDir).exists().isDirectory();
+ assertThat(new File(offsetsDir, "0")).exists().isFile();
+ } finally {
+ query.stop();
+ }
+ }
+
+ /**
+ * A FileIO implementation that rejects operations on paths with a specific prefix. This is used
+ * to test that checkpoint locations don't use the table's FileIO configuration.
+ */
+ public static class RestrictedPathFileIO implements FileIO {
+ private static String rejectedPrefix;
+ private FileIO delegate;
+
+ public static void setRejectedPrefix(String prefix) {
+ rejectedPrefix = prefix;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ validatePath(path);
+ return delegate.newInputFile(path);
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ validatePath(path);
+ return delegate.newOutputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ validatePath(path);
+ delegate.deleteFile(path);
+ }
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ this.delegate = new HadoopFileIO();
+ this.delegate.initialize(properties);
Review Comment:
The `RestrictedPathFileIO.initialize()` delegates to `new HadoopFileIO()` — this means the test itself depends on `HadoopFileIO`. If the goal is to verify checkpoint IO independence from the table FileIO, the test should ideally not hardcode the delegate to be Hadoop-based either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "c2zwdjnlcg (via GitHub)" <gi...@apache.org>.
c2zwdjnlcg commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-3956517094
@RussellSpitzer Thanks for the review.
> please do larger changes like this only on a single module first, then backport to the others in a follow up.
Sorry about that, will keep in mind for next time
Hopefully this is more inline with what you were thinking.
I named the setting `streaming-checkpoint-use-table-io`. If you are generally ok with this approach and name I'll also add documentation to this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2891949434
##########
docs/docs/spark-configuration.md:
##########
@@ -220,6 +220,7 @@ spark.read
| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per microbatch |
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit |
+| streaming-checkpoint-use-hadoop | false | Use Hadoop FileSystem for streaming checkpoint operations instead of the table's FileIO implementation |
Review Comment:
Yeah the only reason why I think we should use HadoopFileIO is that it minimizes the amount of code we have to maintain here. If you are in favor of a vanilla HadoopFS impl I think that's fine too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2891737058
##########
docs/docs/spark-configuration.md:
##########
@@ -220,6 +220,7 @@ spark.read
| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per microbatch |
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit |
+| streaming-checkpoint-use-hadoop | false | Use Hadoop FileSystem for streaming checkpoint operations instead of the table's FileIO implementation |
Review Comment:
If this is purely a spark feature (and it looks like it is), then I would agree that we should just remove and Iceberg/FileIO code path that would complicate how we thing about pathing that's outside the bounds of the Iceberg table and integration.
So if we're going directly through the checkpoint file manager and using hadoop files system natively, that seems like the right "Spark" way to do it. This also avoids adding more spark specific configs to work around the interaction between Iceberg/Spark.
+1 (assuming I understood that correctly).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#issuecomment-3933620338
@c2zwdjnlcg I currently don't have any cycles to review this. Maybe @huaxingao, @RussellSpitzer or @aokolnychyi have some time to review it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2854699612
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -247,52 +267,147 @@ public void prepareForTriggerAvailableNow() {
}
}
- private static class InitialOffsetStore {
+ private abstract static class BaseOffsetStore {
+ private static final Joiner SLASH = Joiner.on("/");
+
private final Table table;
- private final FileIO io;
private final String initialOffsetLocation;
private final Long fromTimestamp;
- InitialOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
+ BaseOffsetStore(Table table, String checkpointLocation, Long fromTimestamp) {
Review Comment:
Could be "long fromTimestamp"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] spark: Don't use table FileIO for checkpointing files [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #15239:
URL: https://github.com/apache/iceberg/pull/15239#discussion_r2886420394
##########
docs/docs/spark-configuration.md:
##########
@@ -220,6 +220,7 @@ spark.read
| stream-from-timestamp | (none) | A timestamp in milliseconds to stream from; if before the oldest known ancestor snapshot, the oldest will be used |
| streaming-max-files-per-micro-batch | INT_MAX | Maximum number of files per microbatch |
| streaming-max-rows-per-micro-batch | INT_MAX | "Soft maximum" number of rows per microbatch; always includes all rows in next unprocessed file, excludes additional files if their inclusion would exceed the soft max limit |
+| streaming-checkpoint-use-hadoop | false | Use Hadoop FileSystem for streaming checkpoint operations instead of the table's FileIO implementation |
Review Comment:
@danielcweeks ^ What do you think about just always using HadoopFS via HadoopFileIO? I think it's clear from the code I linked to that it's what Spark requires to work, so if a user must have configured correctly while TableIO can be anything.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org