xiangfu0 commented on code in PR #19833:
URL: https://github.com/apache/pinot/pull/19833#discussion_r4289850554
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/RowHeapDataBlock.java:
##########
@@ -46,7 +49,11 @@
/// we use [SerializedDataBlock].
public class RowHeapDataBlock implements MseBlock.Data {
private final DataSchema _dataSchema;
- private final List<Object[]> _rows;
+ @Nullable
+ private volatile List<Object[]> _rows;
+ @Nullable
+ private Table _table;
Review Comment:
`_table` is non-volatile but is read/written both in the constructor and
under `synchronized(this)`. That's safe here because the double-checked gate
keys off the volatile `_rows`, and `_table` is never touched on the lock-free
fast path. Worth a one-line comment so a future reader doesn't 'fix' it by
making `_rows` non-volatile.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/RowHeapDataBlock.java:
##########
@@ -162,7 +232,7 @@ public RowHeapDataBlock copyObjectColumns() {
@Override
public SerializedDataBlock asSerialized() {
try {
- return new SerializedDataBlock(DataBlockBuilder.buildFromRows(_rows,
_dataSchema, _aggFunctions));
+ return new SerializedDataBlock(DataBlockBuilder.buildFromRows(getRows(),
_dataSchema, _aggFunctions));
Review Comment:
`asSerialized()` now has a side effect: via `getRows()` it materializes and
caches the List and drops the table reference. That's fine functionally, but
it's a behavior change from the old pure method — if a block is serialized
once, it stops being table-backed afterward. Just flagging so it's intentional.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/RowHeapDataBlock.java:
##########
@@ -76,23 +83,84 @@ public RowHeapDataBlock(List<Object[]> rows, DataSchema
dataSchema,
@Nullable AggregationFunction[] aggFunctions) {
_dataSchema = dataSchema;
_rows = rows;
+ _numRows = rows.size();
+ _aggFunctions = aggFunctions;
+ }
+
+ /// Creates a block that owns a finished table and exposes its existing row
arrays through a cursor.
+ /// The table and its records must not be updated or finished again after
handoff. The table remains reachable
+ /// until this block is released or [#getRows()] materializes the List
fallback.
+ @SuppressWarnings("rawtypes")
+ public RowHeapDataBlock(Table table, DataSchema dataSchema, @Nullable
AggregationFunction[] aggFunctions) {
+ _dataSchema = dataSchema;
+ _table = table;
+ _numRows = table.size();
_aggFunctions = aggFunctions;
}
@Override
public int getNumRows() {
- return _rows.size();
+ List<Object[]> rows = _rows;
+ return rows != null ? rows.size() : _numRows;
}
@Override
public DataSchema getDataSchema() {
return _dataSchema;
}
- /// Returns the rows in the block.
+ /// Returns the rows in the block, materializing and caching a List for
table-backed blocks.
/// The returned list should be considered immutable.
public List<Object[]> getRows() {
- return _rows;
+ List<Object[]> rows = _rows;
+ if (rows != null) {
+ return rows;
+ }
+ synchronized (this) {
+ if (_rows == null) {
+ rows = new ArrayList<>(_numRows);
+ Iterator<Object[]> iterator = getRowIterator();
+ while (iterator.hasNext()) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rows.size(),
"RowHeapDataBlock#getRows");
+ rows.add(iterator.next());
+ }
+ _rows = rows;
+ _table = null;
+ }
+ return _rows;
+ }
+ }
+
+ /// Returns whether rows can still be consumed directly from the owned
table, without a List fallback.
+ public boolean isTableBacked() {
+ return _rows == null;
+ }
+
+ /// Returns a fresh cursor over the owned row arrays. Row arrays and mutable
aggregation results are shared
+ /// with [#getRows()], with the same ownership rules as a block constructed
from a List.
+ public Iterator<Object[]> getRowIterator() {
+ List<Object[]> rows = _rows;
+ if (rows != null) {
+ return rows.iterator();
+ }
+ synchronized (this) {
+ if (_rows != null) {
+ return _rows.iterator();
+ }
+ assert _table != null;
Review Comment:
The table-backed cursor is created under the lock but iterated after the
lock is released. This is only correct because the handed-off table is already
finished and never mutated again, so concurrent read-iteration of
`IndexedTable._topRecords` is safe. That invariant is the load-bearing part of
the contract — good that the constructor Javadoc states it.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java:
##########
@@ -718,11 +719,17 @@ private static Object[] reorderRow(Object[] row, int[]
columnIndices) {
}
private RowHeapDataBlock composeDirectMseBlock(BaseResultsBlock block) {
- List<Object[]> resultRows = block.getRows();
DataSchema inputDataSchema = block.getDataSchema();
- assert resultRows != null && inputDataSchema != null;
+ assert inputDataSchema != null;
ColumnDataType[] inputStoredTypes =
inputDataSchema.getStoredColumnDataTypes();
ColumnDataType[] outputStoredTypes =
_dataSchema.getStoredColumnDataTypes();
+ if (block instanceof GroupByResultsBlock groupByBlock &&
groupByBlock.getTable() != null
Review Comment:
Good guard: the fast table path only triggers when stored types already
match, so no in-place `convertRow` mutation of the live table arrays. When
types differ it correctly falls back to `getRows()`, which for a
`GroupByResultsBlock` already aliases the record arrays — so the in-place
conversion behavior is unchanged from before. Reads correctly to me.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
