This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 9c48dac9d7b Close the posting list stream in OffHeapJsonIndexCreator
(#19925)
9c48dac9d7b is described below
commit 9c48dac9d7ba7d6a9e2086ab32961dfa2a10c254
Author: Deepak kumar <[email protected]>
AuthorDate: Mon Oct 26 10:44:10 2026 -0700
Close the posting list stream in OffHeapJsonIndexCreator (#19925)
---
.../impl/inv/json/OffHeapJsonIndexCreator.java | 25 +++++-
.../segment/local/segment/index/JsonIndexTest.java | 97 ++++++++++++++++++++++
2 files changed, 121 insertions(+), 1 deletion(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/json/OffHeapJsonIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/json/OffHeapJsonIndexCreator.java
index 8234e6ecc4e..526fc116a36 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/json/OffHeapJsonIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/json/OffHeapJsonIndexCreator.java
@@ -74,7 +74,18 @@ public class OffHeapJsonIndexCreator extends
BaseJsonIndexCreator {
throws IOException {
super(indexDir, columnName, tableNameWithType, continueOnError,
jsonIndexConfig);
_postingListFile = new File(_tempDir, POSTING_LIST_FILE_NAME);
- _postingListOutputStream = new DataOutputStream(new
BufferedOutputStream(new FileOutputStream(_postingListFile)));
+ try {
+ _postingListOutputStream = new DataOutputStream(new
BufferedOutputStream(new FileOutputStream(_postingListFile)));
+ } catch (Throwable t) {
+ // The base constructor already created the temp dir, but a failure here
leaves the caller without an instance to
+ // close, so the temp dir has to be cleaned up here. Call super.close()
because close() dereferences the stream.
+ try {
+ super.close();
+ } catch (Throwable cleanupFailure) {
+ t.addSuppressed(cleanupFailure);
+ }
+ throw t;
+ }
}
@Override
@@ -225,6 +236,18 @@ public class OffHeapJsonIndexCreator extends
BaseJsonIndexCreator {
_numPostingLists++;
}
+ @Override
+ public void close()
+ throws IOException {
+ // NOTE: seal() already closes the stream before reading the posting list
file back, but it is not reached when
+ // add() fails. Closing the stream again here is safe because
nothing is written in between.
+ try {
+ _postingListOutputStream.close();
+ } finally {
+ super.close();
+ }
+ }
+
private static class ChunkIterator implements Iterator<PostingListEntry> {
final PinotDataBuffer _dataBuffer;
final long _bufferSize;
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
index 477e0d993a2..193f19351a5 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
@@ -22,8 +22,10 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.StreamReadConstraints;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.google.common.collect.Lists;
+import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.Arrays;
@@ -57,6 +59,7 @@ import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.testng.SkipException;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -1104,6 +1107,100 @@ public class JsonIndexTest implements
PinotBuffersAfterMethodCheckRule {
createIndex(true, jsonIndexConfig, records);
}
+ /// The off-heap creator opens the posting list stream in the constructor,
but seal() is never reached when add()
+ /// fails. close() must still release the stream and clean up the temp
directory, and must stay idempotent so that
+ /// the normal seal()-then-close() path keeps working.
+ @Test
+ public void testOffHeapCreatorCloseWithoutSeal()
+ throws Exception {
+ String columnName = "offHeapUnsealed";
+ File tempDir = new File(INDEX_DIR, columnName + ".json.idx.tmp");
+
+ OffHeapJsonIndexCreator creator =
+ new OffHeapJsonIndexCreator(INDEX_DIR, columnName, "myTable_OFFLINE",
false, getIndexConfig());
+ try {
+ // the braces don't match and cannot be parsed
+ assertThrows(JsonProcessingException.class, () ->
creator.add("{\"key1\":\"va\""));
+ // Sanity check that the probe below can observe an open stream, so that
the assertion after close() cannot pass
+ // vacuously
+ assertTrue(isPostingListStreamOpen(creator));
+ } finally {
+ creator.close();
+ }
+ assertFalse(isPostingListStreamOpen(creator));
+ assertFalse(tempDir.exists());
+
+ // Closing twice must not fail
+ creator.close();
+
+ // The regular path closes the stream in seal() and again in close()
+ OffHeapJsonIndexCreator sealedCreator =
+ new OffHeapJsonIndexCreator(INDEX_DIR, columnName, "myTable_OFFLINE",
false, getIndexConfig());
+ try {
+ sealedCreator.add(TEST_RECORD);
+ sealedCreator.seal();
+ } finally {
+ sealedCreator.close();
+ }
+ assertFalse(isPostingListStreamOpen(sealedCreator));
+ assertTrue(new File(INDEX_DIR, columnName +
V1Constants.Indexes.JSON_INDEX_FILE_EXTENSION).exists());
+ }
+
+ /// The base constructor creates the temp directory before the off-heap
creator opens the posting list stream. When
+ /// opening the stream fails, the caller never gets an instance to close, so
the constructor has to clean up itself.
+ @Test
+ public void testOffHeapCreatorConstructorFailureCleansUpTempDir()
+ throws Exception {
+ String columnName = "offHeapCtorFailure";
+ File tempDir = new File(INDEX_DIR, columnName + ".json.idx.tmp");
+ assertTrue(tempDir.mkdir());
+ try {
+ // Make the posting list file impossible to create
+ tempDir.setWritable(false, false);
+ if (canCreateFileIn(tempDir)) {
+ throw new SkipException("Running as a user that bypasses file
permissions, cannot fail the stream creation");
+ }
+ assertThrows(IOException.class,
+ () -> new OffHeapJsonIndexCreator(INDEX_DIR, columnName,
"myTable_OFFLINE", false, getIndexConfig()));
+ assertFalse(tempDir.exists());
+ } finally {
+ tempDir.setWritable(true, false);
+ FileUtils.deleteQuietly(tempDir);
+ }
+ }
+
+ /// Returns whether a file can be created in the given directory, to detect
environments (e.g. running as root) where
+ /// making a directory read-only does not actually prevent writes.
+ private static boolean canCreateFileIn(File dir) {
+ File probe = new File(dir, "probe");
+ try {
+ if (!probe.createNewFile()) {
+ return false;
+ }
+ } catch (IOException e) {
+ return false;
+ }
+ FileUtils.deleteQuietly(probe);
+ return true;
+ }
+
+ /// Returns whether the posting list stream of the given creator is still
holding its file descriptor, by writing a
+ /// byte and checking whether it reaches the underlying file. The stream is
private with no accessor, and the
+ /// buffering layer accepts writes without touching the file descriptor, so
this needs both reflection and a flush.
+ private static boolean isPostingListStreamOpen(OffHeapJsonIndexCreator
creator)
+ throws Exception {
+ Field field =
OffHeapJsonIndexCreator.class.getDeclaredField("_postingListOutputStream");
+ field.setAccessible(true);
+ DataOutputStream postingListOutputStream = (DataOutputStream)
field.get(creator);
+ try {
+ postingListOutputStream.writeByte(0);
+ postingListOutputStream.flush();
+ return true;
+ } catch (IOException e) {
+ return false;
+ }
+ }
+
@Test
public void testGetMatchingValDocIdsPairForArrayPath() throws Exception {
String[] records = Arrays.asList(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
