Copilot commented on code in PR #9003:
URL: https://github.com/apache/paimon/pull/9003#discussion_r3703944979
##########
paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java:
##########
@@ -597,4 +610,123 @@ public void testReturnedPositionSequential(@TempDir
java.nio.file.Path tempDir)
assertEquals(5, expectedPos, "Should have read exactly 5 rows");
}
}
+
+ @Test
+ public void testReturnedPositionWithMultipleScanPartitions(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("payload", DataTypes.STRING())
+ .build();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(new
Options(), 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile =
+ new Path(new Path(tempDir.toUri()), "test_ordered_scan_" +
UUID.randomUUID());
+
+ // Create multiple scan tasks and make the first one slower, so an
unordered scan returns
+ // a later task first.
+ int firstRangeRowCount = 1_024;
+ int lastSelectedRow = 4_999;
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i <= lastSelectedRow; i++) {
+ String payload = i < firstRangeRowCount ? payload(i) : "x";
+ writer.addElement(GenericRow.of(i,
BinaryString.fromString(payload)));
+ }
+ }
+
+ long[] selectedRows = new long[firstRangeRowCount + 1];
+ List<Integer> expectedIds = new ArrayList<>(selectedRows.length);
+ for (int i = 0; i < firstRangeRowCount; i++) {
+ selectedRows[i] = i;
+ expectedIds.add(i);
+ }
+ selectedRows[firstRangeRowCount] = lastSelectedRow;
+ expectedIds.add(lastSelectedRow);
+
+ int previousWorkerCount = NativeRuntime.workerCount();
+ NativeRuntime.setWorkerThreads(2);
+ try {
+ List<Integer> unorderedIds = readIds(testFile, selectedRows,
false);
+ assertFalse(
+ expectedIds.equals(unorderedIds),
+ "The unordered scan unexpectedly preserved physical row
order");
+ assertEquals(expectedIds, readIds(testFile, selectedRows, true));
Review Comment:
This assertion makes the test non-deterministic: an unordered scan is
allowed to return in physical order, so
`assertFalse(expectedIds.equals(unorderedIds))` can legitimately fail depending
on scheduling/partitioning. To keep the regression test deterministic, drop
this assertion (or make it non-fatal), and only assert that `ordered=true`
matches `expectedIds` (and that `VortexRecordsReader` aligns
`returnedPosition()` accordingly).
##########
paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java:
##########
@@ -597,4 +610,123 @@ public void testReturnedPositionSequential(@TempDir
java.nio.file.Path tempDir)
assertEquals(5, expectedPos, "Should have read exactly 5 rows");
}
}
+
+ @Test
+ public void testReturnedPositionWithMultipleScanPartitions(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("payload", DataTypes.STRING())
+ .build();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(new
Options(), 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile =
+ new Path(new Path(tempDir.toUri()), "test_ordered_scan_" +
UUID.randomUUID());
+
+ // Create multiple scan tasks and make the first one slower, so an
unordered scan returns
+ // a later task first.
+ int firstRangeRowCount = 1_024;
+ int lastSelectedRow = 4_999;
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i <= lastSelectedRow; i++) {
+ String payload = i < firstRangeRowCount ? payload(i) : "x";
+ writer.addElement(GenericRow.of(i,
BinaryString.fromString(payload)));
+ }
+ }
+
+ long[] selectedRows = new long[firstRangeRowCount + 1];
+ List<Integer> expectedIds = new ArrayList<>(selectedRows.length);
+ for (int i = 0; i < firstRangeRowCount; i++) {
+ selectedRows[i] = i;
+ expectedIds.add(i);
+ }
+ selectedRows[firstRangeRowCount] = lastSelectedRow;
+ expectedIds.add(lastSelectedRow);
+
+ int previousWorkerCount = NativeRuntime.workerCount();
+ NativeRuntime.setWorkerThreads(2);
+ try {
+ List<Integer> unorderedIds = readIds(testFile, selectedRows,
false);
+ assertFalse(
+ expectedIds.equals(unorderedIds),
+ "The unordered scan unexpectedly preserved physical row
order");
+ assertEquals(expectedIds, readIds(testFile, selectedRows, true));
+
+ try (VortexRecordsReader reader =
+ new VortexRecordsReader(
+ testFile,
+ rowType,
+ rowType,
+ selectedRows,
+ null,
+ Collections.emptyMap())) {
+ int readCount = 0;
+ FileRecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ InternalRow row;
+ while ((row = batch.next()) != null) {
+ assertEquals(batch.returnedPosition(), row.getInt(0));
+ readCount++;
+ }
+ }
+ assertEquals(selectedRows.length, readCount);
+ }
+ } finally {
+ NativeRuntime.setWorkerThreads(previousWorkerCount);
+ }
Review Comment:
`NativeRuntime.setWorkerThreads(...)` mutates global process state. If the
test suite runs tests concurrently (JUnit 5 parallel execution), this can cause
cross-test interference and flaky failures. Consider isolating this test from
parallel execution (e.g., JUnit 5 `@Execution(SAME_THREAD)` / `@Isolated`)
and/or guarding the worker-thread mutation with a shared lock so only one test
can change it at a time.
##########
paimon-vortex/paimon-vortex-format/src/test/java/org/apache/paimon/format/vortex/VortexReaderWriterTest.java:
##########
@@ -597,4 +610,123 @@ public void testReturnedPositionSequential(@TempDir
java.nio.file.Path tempDir)
assertEquals(5, expectedPos, "Should have read exactly 5 rows");
}
}
+
+ @Test
+ public void testReturnedPositionWithMultipleScanPartitions(@TempDir
java.nio.file.Path tempDir)
+ throws Exception {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("payload", DataTypes.STRING())
+ .build();
+ VortexFileFormat format =
+ new VortexFileFormatFactory()
+ .create(new FileFormatFactory.FormatContext(new
Options(), 1024, 1024));
+
+ FileIO fileIO = new LocalFileIO();
+ Path testFile =
+ new Path(new Path(tempDir.toUri()), "test_ordered_scan_" +
UUID.randomUUID());
+
+ // Create multiple scan tasks and make the first one slower, so an
unordered scan returns
+ // a later task first.
+ int firstRangeRowCount = 1_024;
+ int lastSelectedRow = 4_999;
+ try (FormatWriter writer =
+ ((SupportsDirectWrite) format.createWriterFactory(rowType))
+ .create(fileIO, testFile, "")) {
+ for (int i = 0; i <= lastSelectedRow; i++) {
+ String payload = i < firstRangeRowCount ? payload(i) : "x";
+ writer.addElement(GenericRow.of(i,
BinaryString.fromString(payload)));
+ }
+ }
+
+ long[] selectedRows = new long[firstRangeRowCount + 1];
+ List<Integer> expectedIds = new ArrayList<>(selectedRows.length);
+ for (int i = 0; i < firstRangeRowCount; i++) {
+ selectedRows[i] = i;
+ expectedIds.add(i);
+ }
+ selectedRows[firstRangeRowCount] = lastSelectedRow;
+ expectedIds.add(lastSelectedRow);
+
+ int previousWorkerCount = NativeRuntime.workerCount();
+ NativeRuntime.setWorkerThreads(2);
+ try {
+ List<Integer> unorderedIds = readIds(testFile, selectedRows,
false);
+ assertFalse(
+ expectedIds.equals(unorderedIds),
+ "The unordered scan unexpectedly preserved physical row
order");
+ assertEquals(expectedIds, readIds(testFile, selectedRows, true));
+
+ try (VortexRecordsReader reader =
+ new VortexRecordsReader(
+ testFile,
+ rowType,
+ rowType,
+ selectedRows,
+ null,
+ Collections.emptyMap())) {
+ int readCount = 0;
+ FileRecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ InternalRow row;
+ while ((row = batch.next()) != null) {
+ assertEquals(batch.returnedPosition(), row.getInt(0));
+ readCount++;
+ }
+ }
+ assertEquals(selectedRows.length, readCount);
+ }
+ } finally {
+ NativeRuntime.setWorkerThreads(previousWorkerCount);
+ }
+ }
+
+ private static List<Integer> readIds(Path path, long[] selectedRows,
boolean ordered)
+ throws Exception {
+ List<Integer> ids = new ArrayList<>();
+ BufferAllocator allocator =
+ ArrowAllocation.rootAllocator()
+ .newChildAllocator("vortex-order-test", 0,
Long.MAX_VALUE);
+ try (Session session = Session.create();
+ DataSource dataSource =
+ DataSource.open(session, path.toUri().toString(),
Collections.emptyMap());
+ Scan scan =
+ dataSource.scan(
+ ImmutableScanOptions.builder()
+ .projection(
+ Expression.select(
+ new String[] {"id",
"payload"},
+ Expression.root()))
+ .selectionIndices(selectedRows)
+
.selectionMode(ScanOptions.SelectionMode.INCLUDE)
+ .ordered(ordered)
+ .build())) {
+ while (scan.hasNext()) {
+ try (Partition partition = scan.next();
+ ArrowReader arrowReader =
partition.scanArrow(allocator)) {
+ while (arrowReader.loadNextBatch()) {
+ VectorSchemaRoot root =
arrowReader.getVectorSchemaRoot();
+ IntVector idVector = (IntVector) root.getVector("id");
+ for (int i = 0; i < root.getRowCount(); i++) {
+ ids.add(idVector.get(i));
+ }
+ }
+ }
+ }
+ } finally {
+ allocator.close();
+ }
+ return ids;
+ }
+
+ private static String payload(int rowId) {
+ char[] chars = new char[4_096];
+ int state = rowId + 1;
+ for (int i = 0; i < chars.length; i++) {
+ state = state * 1_103_515_245 + 12_345;
+ chars[i] = (char) ('a' + ((state >>> 16) & 15));
+ }
+ return new String(chars);
+ }
Review Comment:
This helper allocates a 4,096-char payload per call and uses several
unexplained magic constants. It can noticeably increase test runtime/memory,
and its purpose is primarily to influence scan timing. If Comment 1’s
non-deterministic unordered-scan assertion is removed, this can likely be
simplified substantially (smaller payload, fewer rows, or a clearer/cheaper
deterministic fixture). If it must remain, document why `4_096` and the RNG
constants are required to reproduce the behavior.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
