JingsongLi commented on code in PR #9798:
URL: https://github.com/apache/paimon/pull/9798#discussion_r4011477101
##########
paimon-python/pypaimon/globalindex/create_global_index.py:
##########
@@ -288,72 +289,164 @@ def _create_sorted_index_writer(self, index_path: str,
key_serializer):
def _build_generic_index(
self, splits, unindexed_ranges, index_field, table_read, index_path:
str
) -> List[CommitMessage]:
- from pypaimon.read.table_read import _ClosableArrowBatchReader
-
rows_per_shard = self._core_options.global_index_row_count_per_shard()
if rows_per_shard <= 0:
raise ValueError(
"Option 'global-index.row-count-per-shard' must be greater
than 0."
)
- messages = []
- for index_split, index_range in _split_by_global_index_shard(
- splits, rows_per_shard, unindexed_ranges
- ):
- writer = None
+ parallelism = self._core_options.global_index_build_parallelism()
+ if parallelism <= 0:
+ raise ValueError(
+ "Option 'global-index.build.parallelism' must be greater than
0."
+ )
+
+ shards = _split_by_global_index_shard(
+ splits, rows_per_shard, unindexed_ranges)
+ if not shards:
+ return []
+
+ if parallelism == 1 or len(shards) == 1:
+ messages = []
try:
- reader, batches =
table_read._new_arrow_batch_reader([index_split])
- # Close the Python iterator explicitly on failure as well as
- # the Arrow reader, which may retain a suspended generator.
- with _ClosableArrowBatchReader(reader, batches) as
batch_reader:
- for batch in batch_reader:
- if batch.num_rows == 0:
- continue
- if writer is None:
- writer = self._create_generic_index_writer(
- index_path, index_field)
- if self._index_type in VINDEX_IDENTIFIERS:
- if
batch.column(SpecialFields.ROW_ID.name).null_count:
- raise ValueError(
- "Cannot build global index because _ROW_ID
is null.")
- for offset in range(0, batch.num_rows,
ADD_BATCH_SIZE):
- _write_vector_batch(
- writer, batch.slice(offset,
ADD_BATCH_SIZE),
- self._index_columns[0], index_range)
- else:
- for value, row_id in _extract_index_rows(
- batch,
- self._index_columns[0],
- SpecialFields.ROW_ID.name,
- index_range,
- ):
- writer.write(value, row_id - index_range.from_)
- del batch
-
- if writer is None:
+ for index_split, index_range in shards:
+ message = self._build_generic_shard(
+ index_split, index_range, index_field, table_read,
index_path)
+ if message is not None:
+ messages.append(message)
+ return messages
+ except BaseException:
+ self._delete_uncommitted_indexes(messages)
+ raise
+
+ futures = []
+ try:
+ with ThreadPoolExecutor(
+ max_workers=min(parallelism, len(shards)),
+ thread_name_prefix="paimon-global-index-build",
+ ) as executor:
+ futures = [
Review Comment:
[P2] Preserve output ownership when submitting a shard fails
`futures` is assigned only after the entire comprehension succeeds. If
creating a later worker raises `RuntimeError: can't start new thread`, it
remains the empty list from line 322. Previously accepted shards finish during
executor shutdown, but rollback sees no futures and leaves their index files
uncommitted. I reproduced this with a real executor, failure on the second
worker startup and two temporary shard files; both remained after the build
raised. Track completed output ownership independently of successful
construction of the future list and roll it back after workers stop. Merely
appending returned futures is insufficient for this failure: Python enqueues a
work item before starting the additional worker, so even the submission that
raises may run on an existing worker.
--
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]
