You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "bk-mz (via GitHub)" <gi...@apache.org> on 2025/04/21 06:08:39 UTC
[PR] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
bk-mz opened a new pull request, #12856:
URL: https://github.com/apache/iceberg/pull/12856
- Log constructor parameters (branch, case sensitivity, split config, fromTimestamp, max files/records) and the initial offset
- Instrument latestOffset() to log table refresh, START_OFFSET decisions, snapshot IDs, and added file counts
- In planInputPartitions(): log start/end offsets, number of file scan tasks, combined scan tasks, preferred locations, and created partitions
- In planFiles(): trace snapshot loop, current and end file indices, per‑snapshot task counts, and total tasks
- In shouldProcess/nextValidSnapshot(): log skipped snapshots and next valid snapshot IDs
- In the ReadLimit‑aware latestOffset(): log manifest counts, per‑task file/record accumulation, limit checks, snapshot progression, and computed new offset or null
- In addedFilesCount(): log whether count is from summary or by iteration
- In InitialOffsetStore: log reading/writing of offset files, fallback logic, and IO warnings
- Use log.debug for normal tracing and log.warn for IO exception cases
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2052921366
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
+ branch,
+ caseSensitive,
+ splitSize,
+ splitLookback,
+ splitOpenFileCost,
+ fromTimestamp,
+ maxFilesPerMicroBatch,
+ maxRecordsPerMicroBatch);
+
InitialOffsetStore initialOffsetStore =
new InitialOffsetStore(table, checkpointLocation, fromTimestamp);
this.initialOffset = initialOffsetStore.initialOffset();
+ LOG.debug("Initial offset set to {}", initialOffset);
this.skipDelete = readConf.streamingSkipDeleteSnapshots();
this.skipOverwrite = readConf.streamingSkipOverwriteSnapshots();
+ LOG.debug("Skip delete snapshots={}, skip overwrite snapshots={}", skipDelete, skipOverwrite);
}
@Override
public Offset latestOffset() {
table.refresh();
- if (table.currentSnapshot() == null) {
- return StreamingOffset.START_OFFSET;
- }
-
- if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "Refreshed table {}, current snapshot id={}",
+ table.name(),
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : "null");
+
+ if (table.currentSnapshot() == null
+ || table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "No valid current snapshot or snapshot before fromTimestamp ({}), returning START_OFFSET",
+ fromTimestamp);
return StreamingOffset.START_OFFSET;
}
Snapshot latestSnapshot = table.currentSnapshot();
Review Comment:
fixed
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -222,37 +264,74 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
}
Snapshot snapshot = table.snapshot(currentOffset.snapshotId());
-
validateCurrentSnapshotExists(snapshot, currentOffset);
if (!shouldProcess(snapshot)) {
- LOG.debug("Skipping snapshot: {} of table {}", currentOffset.snapshotId(), table.name());
+ LOG.debug(
+ "Skipping processing for snapshot id={} operation={}",
+ snapshot.snapshotId(),
+ snapshot.operation());
continue;
}
- Snapshot currentSnapshot = table.snapshot(currentOffset.snapshotId());
- if (currentOffset.snapshotId() == endOffset.snapshotId()) {
- endFileIndex = endOffset.position();
- } else {
- endFileIndex = addedFilesCount(currentSnapshot);
- }
+ long endFileIndex =
+ currentOffset.snapshotId() == endOffset.snapshotId()
+ ? endOffset.position()
+ : addedFilesCount(snapshot);
MicroBatch latestMicroBatch =
- MicroBatches.from(currentSnapshot, table.io())
+ MicroBatches.from(snapshot, table.io())
.caseSensitive(caseSensitive)
.specsById(table.specs())
.generate(
currentOffset.position(),
endFileIndex,
Long.MAX_VALUE,
currentOffset.shouldScanAllFiles());
-
- fileScanTasks.addAll(latestMicroBatch.tasks());
+ List<FileScanTask> tasks = latestMicroBatch.tasks();
+ Instant snapshotDt = Instant.ofEpochMilli(snapshot.timestampMillis());
+ LOG.debug(
+ "Processing snapshot [id={}, dateTime={}, ageHours={}, startFileIndex={}, endFileIndex={}] generated {} file scan tasks",
+ currentOffset.snapshotId(),
+ DateTimeFormatter.ISO_INSTANT.format(snapshotDt),
+ ChronoUnit.HOURS.between(snapshotDt, Instant.now()),
+ currentOffset.position(),
+ endFileIndex,
+ tasks.size());
+
+ fileScanTasks.addAll(tasks);
} while (currentOffset.snapshotId() != endOffset.snapshotId());
return fileScanTasks;
}
+ public static class FileScanTaskSummary {
+ private long sizeInBytes = 0;
+ private int nrOfFiles = 0;
+ private long evalTimeTakenMs;
+
+ public FileScanTaskSummary(List<FileScanTask> list) {
+ long currentTimeMillis = System.currentTimeMillis();
+ for (var task : list) {
+ nrOfFiles += task.filesCount();
+ sizeInBytes += task.sizeBytes();
+ }
+ this.evalTimeTakenMs = System.currentTimeMillis() - currentTimeMillis;
+ }
Review Comment:
fixed
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-3002220022
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2052921635
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -138,24 +167,32 @@ public InputPartition[] planInputPartitions(Offset start, Offset end) {
"Invalid start offset: %s is not a StreamingOffset",
start);
- if (end.equals(StreamingOffset.START_OFFSET)) {
+ StreamingOffset startOffset = (StreamingOffset) start;
+ StreamingOffset endOffset = (StreamingOffset) end;
+
+ if (endOffset.equals(StreamingOffset.START_OFFSET)) {
+ LOG.debug("End offset is START_OFFSET, returning no partitions");
return new InputPartition[0];
}
- StreamingOffset endOffset = (StreamingOffset) end;
- StreamingOffset startOffset = (StreamingOffset) start;
-
List<FileScanTask> fileScanTasks = planFiles(startOffset, endOffset);
+ FileScanTaskSummary taskSummary = new FileScanTaskSummary(fileScanTasks);
+ LOG.debug(
+ "planFiles returned {} file scan tasks. total_files={}, total_size_in_bytes={}. Time taken to eval stats {} ms",
+ fileScanTasks.size(),
+ taskSummary.nrOfFiles,
+ taskSummary.sizeInBytes,
+ taskSummary.evalTimeTakenMs);
CloseableIterable<FileScanTask> splitTasks =
TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTasks), splitSize);
List<CombinedScanTask> combinedScanTasks =
Lists.newArrayList(
TableScanUtil.planTasks(splitTasks, splitSize, splitLookback, splitOpenFileCost));
- String[][] locations = computePreferredLocations(combinedScanTasks);
+ LOG.debug("Split into {} combined scan tasks", combinedScanTasks.size());
Review Comment:
Can you please elaborate?
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-3029789836
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2823857312
Build is failing with this message:
``` TestSnapshotTableAction > initializationError FAILED
java.io.IOException: Failed to bind to 0.0.0.0/0.0.0.0:32965
at org.eclipse.jetty.server.ServerConnector.openAcceptChannel(ServerConnector.java:344)
at org.eclipse.jetty.server.ServerConnector.open(ServerConnector.java:304)
at org.eclipse.jetty.server.Server.lambda$doStart$0(Server.java:402)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:177)
at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
at org.eclipse.jetty.server.Server.doStart(Server.java:398)
at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:93)
at org.apache.iceberg.rest.RESTCatalogServer.start(RESTCatalogServer.java:116)
at org.apache.iceberg.rest.RESTServerExtension.beforeAll(RESTServerExtension.java:62)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
Caused by:
java.net.BindException: Address already in use
at java.base/sun.nio.ch.Net.bind0(Native Method)
at java.base/sun.nio.ch.Net.bind(Net.java:459)
at java.base/sun.nio.ch.Net.bind(Net.java:448)
at java.base/sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:227)
at org.eclipse.jetty.server.ServerConnector.openAcceptChannel(ServerConnector.java:339)
... 17 more```
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2052647100
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -86,7 +89,7 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
private final int maxFilesPerMicroBatch;
private final int maxRecordsPerMicroBatch;
- SparkMicroBatchStream(
+ public SparkMicroBatchStream(
Review Comment:
why public ?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -222,37 +264,74 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
}
Snapshot snapshot = table.snapshot(currentOffset.snapshotId());
-
validateCurrentSnapshotExists(snapshot, currentOffset);
if (!shouldProcess(snapshot)) {
- LOG.debug("Skipping snapshot: {} of table {}", currentOffset.snapshotId(), table.name());
+ LOG.debug(
+ "Skipping processing for snapshot id={} operation={}",
+ snapshot.snapshotId(),
+ snapshot.operation());
continue;
}
- Snapshot currentSnapshot = table.snapshot(currentOffset.snapshotId());
- if (currentOffset.snapshotId() == endOffset.snapshotId()) {
- endFileIndex = endOffset.position();
- } else {
- endFileIndex = addedFilesCount(currentSnapshot);
- }
+ long endFileIndex =
+ currentOffset.snapshotId() == endOffset.snapshotId()
+ ? endOffset.position()
+ : addedFilesCount(snapshot);
MicroBatch latestMicroBatch =
- MicroBatches.from(currentSnapshot, table.io())
+ MicroBatches.from(snapshot, table.io())
.caseSensitive(caseSensitive)
.specsById(table.specs())
.generate(
currentOffset.position(),
endFileIndex,
Long.MAX_VALUE,
currentOffset.shouldScanAllFiles());
-
- fileScanTasks.addAll(latestMicroBatch.tasks());
+ List<FileScanTask> tasks = latestMicroBatch.tasks();
+ Instant snapshotDt = Instant.ofEpochMilli(snapshot.timestampMillis());
+ LOG.debug(
+ "Processing snapshot [id={}, dateTime={}, ageHours={}, startFileIndex={}, endFileIndex={}] generated {} file scan tasks",
+ currentOffset.snapshotId(),
+ DateTimeFormatter.ISO_INSTANT.format(snapshotDt),
+ ChronoUnit.HOURS.between(snapshotDt, Instant.now()),
+ currentOffset.position(),
+ endFileIndex,
+ tasks.size());
+
+ fileScanTasks.addAll(tasks);
} while (currentOffset.snapshotId() != endOffset.snapshotId());
return fileScanTasks;
}
+ public static class FileScanTaskSummary {
Review Comment:
why public ?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
+ branch,
+ caseSensitive,
+ splitSize,
+ splitLookback,
+ splitOpenFileCost,
+ fromTimestamp,
+ maxFilesPerMicroBatch,
+ maxRecordsPerMicroBatch);
+
InitialOffsetStore initialOffsetStore =
new InitialOffsetStore(table, checkpointLocation, fromTimestamp);
this.initialOffset = initialOffsetStore.initialOffset();
+ LOG.debug("Initial offset set to {}", initialOffset);
this.skipDelete = readConf.streamingSkipDeleteSnapshots();
this.skipOverwrite = readConf.streamingSkipOverwriteSnapshots();
+ LOG.debug("Skip delete snapshots={}, skip overwrite snapshots={}", skipDelete, skipOverwrite);
}
@Override
public Offset latestOffset() {
table.refresh();
- if (table.currentSnapshot() == null) {
- return StreamingOffset.START_OFFSET;
- }
-
- if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "Refreshed table {}, current snapshot id={}",
Review Comment:
Suggestion : I would log both table name and table-uuid
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
Review Comment:
isn't this what we are inferring from the readConf and sparkContext etc ? is this is not easily derviable from the spark UI / history server ?
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
+ branch,
+ caseSensitive,
+ splitSize,
+ splitLookback,
+ splitOpenFileCost,
+ fromTimestamp,
+ maxFilesPerMicroBatch,
+ maxRecordsPerMicroBatch);
+
InitialOffsetStore initialOffsetStore =
new InitialOffsetStore(table, checkpointLocation, fromTimestamp);
this.initialOffset = initialOffsetStore.initialOffset();
+ LOG.debug("Initial offset set to {}", initialOffset);
this.skipDelete = readConf.streamingSkipDeleteSnapshots();
this.skipOverwrite = readConf.streamingSkipOverwriteSnapshots();
+ LOG.debug("Skip delete snapshots={}, skip overwrite snapshots={}", skipDelete, skipOverwrite);
}
@Override
public Offset latestOffset() {
table.refresh();
- if (table.currentSnapshot() == null) {
- return StreamingOffset.START_OFFSET;
- }
-
- if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "Refreshed table {}, current snapshot id={}",
+ table.name(),
+ table.currentSnapshot() != null ? table.currentSnapshot().snapshotId() : "null");
+
+ if (table.currentSnapshot() == null
+ || table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "No valid current snapshot or snapshot before fromTimestamp ({}), returning START_OFFSET",
+ fromTimestamp);
return StreamingOffset.START_OFFSET;
}
Snapshot latestSnapshot = table.currentSnapshot();
Review Comment:
can you move this line above and use latestSnapshot everywhere rather than using table.currentSnapshot()
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -138,24 +167,32 @@ public InputPartition[] planInputPartitions(Offset start, Offset end) {
"Invalid start offset: %s is not a StreamingOffset",
start);
- if (end.equals(StreamingOffset.START_OFFSET)) {
+ StreamingOffset startOffset = (StreamingOffset) start;
+ StreamingOffset endOffset = (StreamingOffset) end;
+
+ if (endOffset.equals(StreamingOffset.START_OFFSET)) {
+ LOG.debug("End offset is START_OFFSET, returning no partitions");
return new InputPartition[0];
}
- StreamingOffset endOffset = (StreamingOffset) end;
- StreamingOffset startOffset = (StreamingOffset) start;
-
List<FileScanTask> fileScanTasks = planFiles(startOffset, endOffset);
+ FileScanTaskSummary taskSummary = new FileScanTaskSummary(fileScanTasks);
+ LOG.debug(
+ "planFiles returned {} file scan tasks. total_files={}, total_size_in_bytes={}. Time taken to eval stats {} ms",
+ fileScanTasks.size(),
+ taskSummary.nrOfFiles,
+ taskSummary.sizeInBytes,
+ taskSummary.evalTimeTakenMs);
CloseableIterable<FileScanTask> splitTasks =
TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTasks), splitSize);
List<CombinedScanTask> combinedScanTasks =
Lists.newArrayList(
TableScanUtil.planTasks(splitTasks, splitSize, splitLookback, splitOpenFileCost));
- String[][] locations = computePreferredLocations(combinedScanTasks);
+ LOG.debug("Split into {} combined scan tasks", combinedScanTasks.size());
Review Comment:
I would use the class name, rather than using the static names as they can change
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -222,37 +264,74 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
}
Snapshot snapshot = table.snapshot(currentOffset.snapshotId());
-
validateCurrentSnapshotExists(snapshot, currentOffset);
if (!shouldProcess(snapshot)) {
- LOG.debug("Skipping snapshot: {} of table {}", currentOffset.snapshotId(), table.name());
+ LOG.debug(
+ "Skipping processing for snapshot id={} operation={}",
+ snapshot.snapshotId(),
+ snapshot.operation());
continue;
}
- Snapshot currentSnapshot = table.snapshot(currentOffset.snapshotId());
- if (currentOffset.snapshotId() == endOffset.snapshotId()) {
- endFileIndex = endOffset.position();
- } else {
- endFileIndex = addedFilesCount(currentSnapshot);
- }
+ long endFileIndex =
+ currentOffset.snapshotId() == endOffset.snapshotId()
+ ? endOffset.position()
+ : addedFilesCount(snapshot);
MicroBatch latestMicroBatch =
- MicroBatches.from(currentSnapshot, table.io())
+ MicroBatches.from(snapshot, table.io())
.caseSensitive(caseSensitive)
.specsById(table.specs())
.generate(
currentOffset.position(),
endFileIndex,
Long.MAX_VALUE,
currentOffset.shouldScanAllFiles());
-
- fileScanTasks.addAll(latestMicroBatch.tasks());
+ List<FileScanTask> tasks = latestMicroBatch.tasks();
+ Instant snapshotDt = Instant.ofEpochMilli(snapshot.timestampMillis());
+ LOG.debug(
+ "Processing snapshot [id={}, dateTime={}, ageHours={}, startFileIndex={}, endFileIndex={}] generated {} file scan tasks",
+ currentOffset.snapshotId(),
+ DateTimeFormatter.ISO_INSTANT.format(snapshotDt),
+ ChronoUnit.HOURS.between(snapshotDt, Instant.now()),
+ currentOffset.position(),
+ endFileIndex,
+ tasks.size());
+
+ fileScanTasks.addAll(tasks);
} while (currentOffset.snapshotId() != endOffset.snapshotId());
return fileScanTasks;
}
+ public static class FileScanTaskSummary {
+ private long sizeInBytes = 0;
+ private int nrOfFiles = 0;
+ private long evalTimeTakenMs;
+
+ public FileScanTaskSummary(List<FileScanTask> list) {
+ long currentTimeMillis = System.currentTimeMillis();
+ for (var task : list) {
+ nrOfFiles += task.filesCount();
+ sizeInBytes += task.sizeBytes();
+ }
+ this.evalTimeTakenMs = System.currentTimeMillis() - currentTimeMillis;
+ }
Review Comment:
can you please move this out of the constructor and to it's individual functions and cache it ? we need this info only debug logging is enabled and it might be a bit expensive.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2052733653
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
Review Comment:
Unfortunately, I couldn't find anything related to that from spark ui.
Microbatch Scan:
<img width="1058" alt="image" src="https://github.com/user-attachments/assets/07ecb82c-5f76-43f9-afcd-47d717f3c0b7" />
```text
== Parsed Logical Plan ==
WriteToMicroBatchDataSourceV1 ForeachBatchSink, d053491a-73d0-4d8b-b364-ecef987788b9, [checkpointLocation=s3://my-table/checkpoints7, fanout-enabled=true], Append, 26
+- StreamingDataSourceV2Relation [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6, headers#7], IcebergScan(table=my_catalog.my_database.my_table, branch=null, type=struct<1: key: optional binary, 2: value: optional binary, 3: topic: optional string, 4: partition: optional int, 5: offset: optional long, 6: timestamp: optional timestamptz, 7: timestampType: optional int, 8: headers: optional list<struct<10: key: optional string, 11: value: optional binary>>>, filters=[], runtimeFilters=[], caseSensitive=false), org.apache.iceberg.spark.source.SparkMicroBatchStream@4b3c3e87, Streaming Offset[7176483676499627176: position (0) scan_all_files (false)], Streaming Offset[3260915119082769976: position (3) scan_all_files (false)]
== Analyzed Logical Plan ==
key: binary, value: binary, topic: string, partition: int, offset: bigint, timestamp: timestamp, timestampType: int, headers: array<struct<key:string,value:binary>>
WriteToMicroBatchDataSourceV1 ForeachBatchSink, d053491a-73d0-4d8b-b364-ecef987788b9, [checkpointLocation=s3://my-table/checkpoints7, fanout-enabled=true], Append, 26
+- StreamingDataSourceV2Relation [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6, headers#7], IcebergScan(table=my_catalog.my_database.my_table, branch=null, type=struct<1: key: optional binary, 2: value: optional binary, 3: topic: optional string, 4: partition: optional int, 5: offset: optional long, 6: timestamp: optional timestamptz, 7: timestampType: optional int, 8: headers: optional list<struct<10: key: optional string, 11: value: optional binary>>>, filters=[], runtimeFilters=[], caseSensitive=false), org.apache.iceberg.spark.source.SparkMicroBatchStream@4b3c3e87, Streaming Offset[7176483676499627176: position (0) scan_all_files (false)], Streaming Offset[3260915119082769976: position (3) scan_all_files (false)]
== Optimized Logical Plan ==
StreamingDataSourceV2Relation [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6, headers#7], IcebergScan(table=my_catalog.my_database.my_table, branch=null, type=struct<1: key: optional binary, 2: value: optional binary, 3: topic: optional string, 4: partition: optional int, 5: offset: optional long, 6: timestamp: optional timestamptz, 7: timestampType: optional int, 8: headers: optional list<struct<10: key: optional string, 11: value: optional binary>>>, filters=[], runtimeFilters=[], caseSensitive=false), org.apache.iceberg.spark.source.SparkMicroBatchStream@4b3c3e87, Streaming Offset[7176483676499627176: position (0) scan_all_files (false)], Streaming Offset[3260915119082769976: position (3) scan_all_files (false)]
== Physical Plan ==
*(1) Project [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6, headers#7]
+- MicroBatchScan[key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6, headers#7] my_catalog.my_database.my_table (branch=null) [filters=, groupedBy=]
```
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2819031939
I'll take a look on above.
JIC, logs from `latestOffset`
```text
2025-04-21 14:04:36.643 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - latestOffset(startOffset=Streaming Offset[3821473473156059401: position (24) scan_all_files (false)], limit=MaxRows: 50000000) called
2025-04-21 14:04:36.857 INFO [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.BaseMetastoreTableOperations [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Refreshing table metadata from new version: s3://my-table/table/metadata/15032-bd9347a5-4e80-4073-803d-6f0e4f8e14f0.metadata.json
2025-04-21 14:04:37.061 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Effective startingOffset=Streaming Offset[3821473473156059401: position (24) scan_all_files (false)]
2025-04-21 14:04:37.062 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Start position in snapshot=24, scanAllFiles=false
2025-04-21 14:04:37.106 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Snapshot 3821473473156059401 has 1 manifest files after skipping
2025-04-21 14:04:37.106 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Reading manifest s3://my-table/table/metadata/6c61afec-a595-4f22-898c-8b3eb71a670c-m0.avro at index 0
2025-04-21 14:04:37.153 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Next valid snapshot is 2193338683398121123
2025-04-21 14:04:37.153 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Moving to next snapshot 2193338683398121123, scanAllFiles reset to false
2025-04-21 14:04:37.193 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Snapshot 2193338683398121123 has 1 manifest files after skipping
2025-04-21 14:04:37.193 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Reading manifest s3://my-table/table/metadata/3ab6f7f5-569d-409d-8c17-a4bbd397e522-m0.avro at index 0
...
2025-04-21 14:04:38.006 DEBUG [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Limits reached: filesAdded=175 (max=2147483647), recordsCount=49840537 (max=50000000), stopping
2025-04-21 14:04:38.006 INFO [ip-172-31-8-223.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = a33ebf30-ab06-4354-8cdc-b0f2afa16849]] - Computed next streaming offset [offset=Streaming Offset[1601835420991821798: position (10) scan_all_files (false)]] after filling the batch: [files=175, records=49840537, bytes=41418549762, snapshots=11]
```
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2052921152
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -222,37 +264,74 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
}
Snapshot snapshot = table.snapshot(currentOffset.snapshotId());
-
validateCurrentSnapshotExists(snapshot, currentOffset);
if (!shouldProcess(snapshot)) {
- LOG.debug("Skipping snapshot: {} of table {}", currentOffset.snapshotId(), table.name());
+ LOG.debug(
+ "Skipping processing for snapshot id={} operation={}",
+ snapshot.snapshotId(),
+ snapshot.operation());
continue;
}
- Snapshot currentSnapshot = table.snapshot(currentOffset.snapshotId());
- if (currentOffset.snapshotId() == endOffset.snapshotId()) {
- endFileIndex = endOffset.position();
- } else {
- endFileIndex = addedFilesCount(currentSnapshot);
- }
+ long endFileIndex =
+ currentOffset.snapshotId() == endOffset.snapshotId()
+ ? endOffset.position()
+ : addedFilesCount(snapshot);
MicroBatch latestMicroBatch =
- MicroBatches.from(currentSnapshot, table.io())
+ MicroBatches.from(snapshot, table.io())
.caseSensitive(caseSensitive)
.specsById(table.specs())
.generate(
currentOffset.position(),
endFileIndex,
Long.MAX_VALUE,
currentOffset.shouldScanAllFiles());
-
- fileScanTasks.addAll(latestMicroBatch.tasks());
+ List<FileScanTask> tasks = latestMicroBatch.tasks();
+ Instant snapshotDt = Instant.ofEpochMilli(snapshot.timestampMillis());
+ LOG.debug(
+ "Processing snapshot [id={}, dateTime={}, ageHours={}, startFileIndex={}, endFileIndex={}] generated {} file scan tasks",
+ currentOffset.snapshotId(),
+ DateTimeFormatter.ISO_INSTANT.format(snapshotDt),
+ ChronoUnit.HOURS.between(snapshotDt, Instant.now()),
+ currentOffset.position(),
+ endFileIndex,
+ tasks.size());
+
+ fileScanTasks.addAll(tasks);
} while (currentOffset.snapshotId() != endOffset.snapshotId());
return fileScanTasks;
}
+ public static class FileScanTaskSummary {
Review Comment:
fixed
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -86,7 +89,7 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
private final int maxFilesPerMicroBatch;
private final int maxRecordsPerMicroBatch;
- SparkMicroBatchStream(
+ public SparkMicroBatchStream(
Review Comment:
fixed
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
+ branch,
+ caseSensitive,
+ splitSize,
+ splitLookback,
+ splitOpenFileCost,
+ fromTimestamp,
+ maxFilesPerMicroBatch,
+ maxRecordsPerMicroBatch);
+
InitialOffsetStore initialOffsetStore =
new InitialOffsetStore(table, checkpointLocation, fromTimestamp);
this.initialOffset = initialOffsetStore.initialOffset();
+ LOG.debug("Initial offset set to {}", initialOffset);
this.skipDelete = readConf.streamingSkipDeleteSnapshots();
this.skipOverwrite = readConf.streamingSkipOverwriteSnapshots();
+ LOG.debug("Skip delete snapshots={}, skip overwrite snapshots={}", skipDelete, skipOverwrite);
}
@Override
public Offset latestOffset() {
table.refresh();
- if (table.currentSnapshot() == null) {
- return StreamingOffset.START_OFFSET;
- }
-
- if (table.currentSnapshot().timestampMillis() < fromTimestamp) {
+ LOG.debug(
+ "Refreshed table {}, current snapshot id={}",
Review Comment:
fixed
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2820640387
> Thanks for adding this verbose logging ! This is gonna be really helpful. I am also parallely working on fixing some of the streaming limitation here ! [1] stream get stuck [2] Trigger Available now : https://issues.apache.org/jira/browse/SPARK-36533
@singhpk234 can you elaborate about [1] stream get stuck?
Do you have an issue opened or more data on 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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2823342898
> its not of a bug but more of a limitation
@singhpk234 I added warn about this here: https://github.com/apache/iceberg/pull/12856/files#diff-b4fbb13d5a4c4bc603648af8f9da3a1ffaf0a51a0708fde0abc2c5c85accc97cR443-R451
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2055799165
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
Review Comment:
>or we need this along the code for quick reference ?
this.
>mm, so I am assuming there was no override of the defaults then wouldn't whats there suff
i was overrriding max rows per batch, but it's not shown in sparkui.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
Review Comment:
>or we need this along the code for quick reference ?
this.
>mm, so I am assuming there was no override of the defaults then wouldn't whats there suff
i was overrriding max rows per batch, but it's not shown in sparkui.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2817719807
@singhpk234 👋 can you look into this as well? 🙏
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #12856: Add detailed debug and warn logging to SparkMicroBatchStream
URL: https://github.com/apache/iceberg/pull/12856
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2054452974
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -138,24 +167,32 @@ public InputPartition[] planInputPartitions(Offset start, Offset end) {
"Invalid start offset: %s is not a StreamingOffset",
start);
- if (end.equals(StreamingOffset.START_OFFSET)) {
+ StreamingOffset startOffset = (StreamingOffset) start;
+ StreamingOffset endOffset = (StreamingOffset) end;
+
+ if (endOffset.equals(StreamingOffset.START_OFFSET)) {
+ LOG.debug("End offset is START_OFFSET, returning no partitions");
return new InputPartition[0];
}
- StreamingOffset endOffset = (StreamingOffset) end;
- StreamingOffset startOffset = (StreamingOffset) start;
-
List<FileScanTask> fileScanTasks = planFiles(startOffset, endOffset);
+ FileScanTaskSummary taskSummary = new FileScanTaskSummary(fileScanTasks);
+ LOG.debug(
+ "planFiles returned {} file scan tasks. total_files={}, total_size_in_bytes={}. Time taken to eval stats {} ms",
+ fileScanTasks.size(),
+ taskSummary.nrOfFiles,
+ taskSummary.sizeInBytes,
+ taskSummary.evalTimeTakenMs);
CloseableIterable<FileScanTask> splitTasks =
TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTasks), splitSize);
List<CombinedScanTask> combinedScanTasks =
Lists.newArrayList(
TableScanUtil.planTasks(splitTasks, splitSize, splitLookback, splitOpenFileCost));
- String[][] locations = computePreferredLocations(combinedScanTasks);
+ LOG.debug("Split into {} combined scan tasks", combinedScanTasks.size());
Review Comment:
ya was asking to use the `CombinedScanTask.class.getName()` to make it future proof to any scan task refactor.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#discussion_r2054458183
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java:
##########
@@ -105,28 +108,54 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
this.maxFilesPerMicroBatch = readConf.maxFilesPerMicroBatch();
this.maxRecordsPerMicroBatch = readConf.maxRecordsPerMicroBatch();
+ LOG.info(
+ "Initializing SparkMicroBatchStream with params: branch={}, caseSensitive={}, "
+ + "splitSize={}, splitLookback={}, splitOpenFileCost={}, fromTimestamp={}, "
+ + "maxFilesPerMicroBatch={}, maxRecordsPerMicroBatch={}",
Review Comment:
Hmm, so I am assuming there was no override of the defaults then wouldn't whats there suff ? or we need this along the code for quick reference ?
I think I am fine with either, since this is debug.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2821859083
@bk-mz its not of a bug but more of a limitation : https://github.com/apache/iceberg/pull/12217#discussion_r1962211721
please ref this ^^
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2818352166
These changes logs examples from planFiles:
```txt
2025-04-21 12:43:04.411 INFO [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Initializing SparkMicroBatchStream with params: branch=null, caseSensitive=false, splitSize=134217728, splitLookback=10, splitOpenFileCost=4194304, fromTimestamp=1745239382655, maxFilesPerMicroBatch=2147483647, maxRecordsPerMicroBatch=50000000
2025-04-21 12:43:04.412 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - InitialOffsetStore created with location s3://com.twilio.mdp.iceberg.tables.mdr.finalized.testcell/checkpoints7/sources/0/offsets/0
2025-04-21 12:43:04.428 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Found existing offset file at s3://com.twilio.mdp.iceberg.tables.mdr.finalized.testcell/checkpoints7/sources/0/offsets/0, reading
2025-04-21 12:43:04.444 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Read offset Streaming Offset[-1: position (-1) scan_all_files (false)] from s3://com.twilio.mdp.iceberg.tables.mdr.finalized.testcell/checkpoints7/sources/0/offsets/0
2025-04-21 12:43:04.444 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Initial offset set to Streaming Offset[-1: position (-1) scan_all_files (false)]
2025-04-21 12:43:04.444 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Skip delete snapshots=true, skip overwrite snapshots=true
2025-04-21 12:43:04.806 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Deserialized offset from JSON {"version":1,"snapshot_id":1823134505898519413,"position":8,"scan_all_files":false}: Streaming Offset[1823134505898519413: position (8) scan_all_files (false)]
2025-04-21 12:43:04.806 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Deserialized offset from JSON {"version":1,"snapshot_id":3821473473156059401,"position":24,"scan_all_files":false}: Streaming Offset[3821473473156059401: position (24) scan_all_files (false)]
2025-04-21 12:43:04.867 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.aws.glue.GlueCatalog [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Using optimistic locking for Glue Data Catalog tables.
2025-04-21 12:43:05.152 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=1823134505898519413, dateTime=2025-04-19T20:53:29.457Z, ageHours=39, startFileIndex=8, endFileIndex=8] generated 0 file scan tasks
2025-04-21 12:43:05.265 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=267320482196197876, dateTime=2025-04-19T20:54:28.367Z, ageHours=39, startFileIndex=0, endFileIndex=8] generated 8 file scan tasks
2025-04-21 12:43:05.266 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Skipping processing for snapshot id=7905888728843236371 operation=replace
2025-04-21 12:43:05.352 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=7421363600347898867, dateTime=2025-04-19T20:55:33.095Z, ageHours=39, startFileIndex=0, endFileIndex=8] generated 8 file scan tasks
2025-04-21 12:43:05.471 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=1672029328327466770, dateTime=2025-04-19T20:56:31.462Z, ageHours=39, startFileIndex=0, endFileIndex=9] generated 9 file scan tasks
2025-04-21 12:43:05.561 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=7190315158754434941, dateTime=2025-04-19T20:57:26.865Z, ageHours=39, startFileIndex=0, endFileIndex=8] generated 8 file scan tasks
...
2025-04-21 12:43:11.720 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Processing snapshot [id=3821473473156059401, dateTime=2025-04-19T22:01:46.828Z, ageHours=38, startFileIndex=0, endFileIndex=24] generated 24 file scan tasks
2025-04-21 12:43:11.721 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - planFiles returned 764 file scan tasks. total_files=764, total_size_in_bytes=178383916431. Time taken to eval stats 0 ms
2025-04-21 12:43:11.734 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Split into 1484 combined scan tasks
2025-04-21 12:43:11.736 DEBUG [ip-0-0-0-0.ec2.internal] - org.apache.iceberg.spark.source.SparkMicroBatchStream [stream execution thread for [id = d053491a-73d0-4d8b-b364-ecef987788b9, runId = 1a1a6d0c-045b-443e-94e3-7c39977f067b]] - Created 1484 SparkInputPartitions
```
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2906077159
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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] Add detailed debug and warn logging to SparkMicroBatchStream [iceberg]
Posted by "bk-mz (via GitHub)" <gi...@apache.org>.
bk-mz commented on PR #12856:
URL: https://github.com/apache/iceberg/pull/12856#issuecomment-2906515915
@singhpk234 Hello again. Do you think it's worth merging this PR? Anything you'd like to change maybe?
--
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