[MSE] Add Apache Arrow as native columnar block format

[praveenc7]

Summary

#18205

Testing Done

TODO

[codecov-commenter]

Codecov Report

❌ Patch coverage is 33.02836% with 732 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.26%. Comparing base (9ff14da) to head (be4bb9f).
⚠️ Report is 9 commits behind head on master.

Files with missing lines	Patch %	Lines
.../apache/pinot/common/datablock/ArrowDataBlock.java	17.39%	126 Missing and 7 partials ⚠️
...inot/query/runtime/blocks/ArrowBlockConverter.java	29.62%	86 Missing and 9 partials ⚠️
...not/query/mailbox/flight/FlightSendingMailbox.java	0.00%	77 Missing ⚠️
...inot/query/mailbox/flight/FlightMailboxServer.java	0.00%	73 Missing ⚠️
...ot/query/runtime/operator/join/ArrowJoinProbe.java	61.58%	55 Missing and 13 partials ⚠️
.../planner/partitioning/SingleColumnKeySelector.java	21.42%	59 Missing and 7 partials ⚠️
.../query/runtime/operator/join/ArrowLookupTable.java	54.54%	49 Missing and 11 partials ⚠️
.../apache/pinot/segment/spi/memory/ArrowBuffers.java	0.00%	32 Missing ⚠️
.../apache/pinot/query/runtime/blocks/ArrowBlock.java	50.98%	22 Missing and 3 partials ⚠️
...pinot/query/runtime/operator/HashJoinOperator.java	63.76%	15 Missing and 10 partials ⚠️
... and 14 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18207      +/-   ##
============================================
- Coverage     63.34%   63.26%   -0.09%     
+ Complexity     1627     1624       -3     
============================================
  Files          3238     3252      +14     
  Lines        197003   198121    +1118     
  Branches      30464    30644     +180     
============================================
+ Hits         124801   125349     +548     
- Misses        62202    62670     +468     
- Partials      10000    10102     +102     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-11	63.24% <33.02%> (+7.99%)	⬆️
java-21	34.74% <0.00%> (-28.58%)	⬇️
temurin	63.26% <33.02%> (-0.09%)	⬇️
unittests	63.26% <33.02%> (-0.09%)	⬇️
unittests1	55.14% <33.02%> (-0.15%)	⬇️
unittests2	34.84% <0.00%> (-0.12%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[yashmayya]

Code review

Found 10 issues:

1. SENTINEL collision causes silent data loss for Integer.MIN_VALUE join keys. IntKeyHasher.computeHash() returns the raw int value, so keys equal to Integer.MIN_VALUE produce a hash identical to the empty-slot sentinel. The probe loop in ArrowJoinProbe.collectMatchesWithCollisions() breaks on sentinel, so these rows are silently dropped in inner joins or treated as unmatched in outer joins.

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/join/ArrowLookupTable.java

Lines 53 to 55 in 982ffa1

	/** Sentinel value in the hash table marking an empty slot. */
	public static final int SENTINEL = Integer.MIN_VALUE;

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/join/ArrowJoinProbe.java

Lines 309 to 313 in 982ffa1

	while (true) {
	int rightHash = _hashes[slot];
	if (rightHash == ArrowLookupTable.SENTINEL) {
	break;
	}

2. NPE in ArrowLookupTable.finalize() when the right side is empty. mergeBlocks() returns null for an empty batch list, then _keySelector.getArrowHasher(null) is called before the null guard in buildHashTable().

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/join/ArrowLookupTable.java

Lines 120 to 124 in 982ffa1

	*/
	public void finalize() {
	_mergedBlock = mergeBlocks(_rightBatches);
	_keyHasher = _keySelector.getArrowHasher(_mergedBlock);

3. RIGHT / FULL OUTER JOIN broken in Arrow mode. When needUnmatchedRightRows() is true, buildNonMatchRightRows() reads _rightTable which is never populated in the Arrow path (only _arrowRightTable is set). The assert on _rightTable != null will fire under -ea; without it, unmatched right rows are silently lost. The TODO at line 256 acknowledges this gap.

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/HashJoinOperator.java

Lines 254 to 258 in 982ffa1

	}
	if (needUnmatchedRightRows()) {
	// TODO: emit unmatched right rows as Arrow blocks
	List<Object[]> rows = buildNonMatchRightRows();
	if (!rows.isEmpty()) {

4. Hardcoded dictionary ID 0 for all dictionary-encoded columns. dictionaryEncoded() and listOfDictionaryEncoded() both use new DictionaryEncoding(0, true, null). Schemas with multiple STRING/JSON columns will share a single dictionary ID, causing DictionaryProvider.lookup() to return the wrong dictionary for the second and subsequent columns -- silent data corruption.

pinot/pinot-common/src/main/java/org/apache/pinot/common/arrow/ArrowSchemaConverter.java

Lines 155 to 158 in 982ffa1

	private static Field dictionaryEncoded(String name) {
	DictionaryEncoding encoding = new DictionaryEncoding(0, true, null);
	FieldType indexFieldType = new FieldType(true, new ArrowType.Int(32, true), encoding, null);
	return new Field(name, indexFieldType, null);

5. arrowTypeToPinot() maps all ArrowType.List to STRING_ARRAY. The reverse conversion ignores the list element type, so INT_ARRAY, LONG_ARRAY, FLOAT_ARRAY, DOUBLE_ARRAY, BOOLEAN_ARRAY, and BYTES_ARRAY are all lossily mapped to STRING_ARRAY on round-trip.

pinot/pinot-common/src/main/java/org/apache/pinot/common/arrow/ArrowSchemaConverter.java

Lines 95 to 98 in 982ffa1

	}
	if (arrowType instanceof ArrowType.List) {
	return ColumnDataType.STRING_ARRAY;
	}

6. Child allocator leaked per Flight batch in FlightMailboxService.processStream(). A new BufferAllocator is created per received block but never closed. Closing VectorSchemaRoot releases buffers but does not close the allocator itself, preventing the root allocator from shutting down cleanly.

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/flight/FlightMailboxService.java

Lines 198 to 201 in 982ffa1

	try (ArrowRecordBatch batch = unloader.getRecordBatch()) {
	BufferAllocator blockAllocator = _arrowBuffers.newQueryAllocator("flight-stream-block");
	VectorSchemaRoot ownedRoot = VectorSchemaRoot.create(root.getSchema(), blockAllocator);
	VectorLoader loader = new VectorLoader(ownedRoot);

7. ArrowBlock not released after Flight send in FlightSendingMailbox.sendDataBlock(). The ArrowBlock obtained via data.asArrow() is used for transfer but release() is never called on the normal path, leaking off-heap memory per sent block. (The early-termination path at line 93 does call release().)

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/flight/FlightSendingMailbox.java

Lines 113 to 118 in 982ffa1

	private void sendDataBlock(MseBlock.Data data) {
	ArrowBlock arrowBlock = data.asArrow();
	VectorSchemaRoot root = arrowBlock.getDataBlock().getRoot();
	DictionaryProvider dictionaryProvider = arrowBlock.getDataBlock().getDictionaryProvider();

8. Race condition in ArrowBlock.retain(). The check-then-act on _refCount is not atomic: between _refCount.get() == 0 and _refCount.incrementAndGet(), another thread can release() to 0 and close the underlying buffers. A CAS loop is needed.

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ArrowBlock.java

Lines 145 to 151 in 982ffa1

	/** Increments the reference count. Must be paired with a subsequent {@link #release()}. */
	public void retain() {
	if (_refCount.get() == 0) {
	throw new IllegalStateException("ArrowBlock has already been released");
	}
	_refCount.incrementAndGet();
	}

9. ArrowBlock.isSerialized() returns true incorrectly. ArrowBlock.isRowHeap() returns false, so the inherited default isSerialized() (!isRowHeap()) returns true. QueryDispatcher branches on isSerialized() and will route ArrowBlock through reduceSerialized(), triggering an expensive and unintended Arrow -> RowHeap -> Serialized double conversion. ArrowBlock should override isSerialized() to return false.

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/MseBlock.java

Lines 107 to 110 in 982ffa1

	/// Returns whether the block is a [SerializedDataBlock].
	default boolean isSerialized() {
	return !isRowHeap();
	}

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ArrowBlock.java

Lines 74 to 77 in 982ffa1

	@Override
	public boolean isRowHeap() {
	return false;
	}

10. New public inner classes in SingleColumnKeySelector lack class-level Javadoc. 9 new public static inner classes (IntKeyHasher, LongKeyHasher, DictionaryKeyHasher, GenericKeyHasher, IntKeyComparator, LongKeyComparator, BothDictionaryKeyComparator, LeftDictionaryKeyComparator, GenericKeyComparator) have no Javadoc. (CLAUDE.md says "Add class-level Javadoc for new classes; describe behavior and thread-safety")

pinot/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/partitioning/SingleColumnKeySelector.java

Lines 118 to 121 in 982ffa1

	public static class IntKeyHasher implements ArrowKeyHasher {
	private final IntVector _vector;

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[praveenc7]

@yashmayya Still working on this, will let you known by next week. When it is ready for full review. I am testing different stratergy to see the performance difference