xiangfu0 commented on code in PR #19833:
URL: https://github.com/apache/pinot/pull/19833#discussion_r4278286600
##########
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() {
Review Comment:
Minor naming: `isTableBacked()` effectively answers "have the rows not been
materialized into the List fallback yet?" — it flips to false after
`getRows()`/`asSerialized()` even for a table-constructed block. Since this is
public API the paired consumer depends on, a name or Javadoc that makes the
"not-yet-materialized" meaning explicit would read more clearly.
##########
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() {
Review Comment:
Nit: `getRowIterator()` re-calls `_table.iterator()` on every invocation and
never memoizes, so repeated calls re-iterate the table rather than reusing a
cursor. That's fine for today's single-pass consumers, but a short note that it
does not cache (unlike `getRows()`) would help a future caller avoid assuming
it's a cheap repeatable cursor.
##########
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 `assert _table != null` leans on the invariant that exactly one of
`_rows`/`_table` is set at construction. That holds today; just noting an
`IllegalStateException` would fail louder than a disabled assertion if a future
constructor ever breaks the invariant. Optional.
--
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]
